Envision VM (part 3), Atomes et Stockage de données

Victor Nicollet

Cet article est le troisième d’une série en quatre parties sur le fonctionnement interne de la machine virtuelle Envision : le logiciel qui exécute les scripts Envision. Voir partie 1, partie 2 et partie 4. Cette série ne couvre pas le compilateur Envision (peut-être une autre fois), alors supposons simplement que le script a d’une manière ou d’une autre été converti en bytecode que la machine virtuelle Envision accepte en entrée.

Pendant l’exécution, les thunks lisent des données d’entrée et écrivent des données de sortie, souvent en grandes quantités.

• Un milliard de booléens (un bit par valeur) occupent 125MB.
• Un milliard de nombres à virgule flottante (précision 32 bits) occupent 4GB.
• Un milliard de lignes de ventes minimales (date, lieu, EAN-13, quantité) occupent entre 14GB et 33GB (voire plus !) selon la manière dont les valeurs sont encodées.

Cela pose deux défis : comment préserver ces données dès leur création et jusqu’à leur utilisation (une partie de la réponse réside dans l’utilisation de disques NVMe répartis sur plusieurs machines), et comment minimiser la quantité de données transitant par des canaux plus lents que la RAM (réseau et stockage persistant).

Couche de métadonnées

Une partie de la solution consiste à avoir deux couches de données distinctes, la donnée étant dirigée vers l’une ou l’autre couche en fonction de sa nature. La couche de métadonnées contient des informations sur les données réelles, ainsi que sur les scripts en cours d’exécution:

• Lorsqu’un thunk a renvoyé avec succès des données, l’identifiant unique de ces données est conservé dans cette couche.
• Lorsqu’un thunk a échoué, les messages d’erreur produits par le thunk sont conservés dans cette couche.
• Lorsqu’un thunk a renvoyé un nouveau thunk (et le DAG de ses parents), le DAG sérialisé est conservé dans cette couche.
• Un thunk peut sauvegarder des points de contrôle dans la couche de métadonnées (généralement constitués de l’identifiant d’un bloc de données); si un thunk est interrompu avant d’être terminé, il peut alors recharger son point de contrôle depuis la couche de métadonnées et reprendre le travail à partir de cette position.

Autrement dit, la couche de métadonnées peut être vue comme un dictionnaire qui associe des thunks à des résultats, la nature exacte du résultat dépendant de ce que le thunk a réellement renvoyé.

La couche de métadonnées peut également contenir des informations supplémentaires concernant la structure des données référencées. Par exemple, si un thunk a renvoyé une paire de vecteurs, alors les métadonnées contiendront l’identifiant unique de chaque vecteur. Cela permet aux consommateurs d’accéder à un vecteur sans avoir à charger les deux.

Il existe deux limites pour les valeurs stockées dans la couche de métadonnées : une entrée ne peut pas dépasser 10MB (donc un DAG sérialisé n’est pas autorisé à dépasser cette taille non plus !), et l’espace de stockage total pour la couche de métadonnées est de 1.5GB. En général, il y a environ un million de valeurs dans cette couche, avec une taille moyenne par entrée de 1.5KB.

La couche de métadonnées réside toujours en RAM pour garantir un accès rapide. Elle sert de source de vérité pour l’exécution des thunks : un thunk a été exécuté si, et seulement si, un résultat lui est associé dans la couche de métadonnées — bien que cela ne garantisse pas que les données référencées par ce résultat soient disponibles.

Chaque worker dans le cluster conserve sa propre copie de la couche de métadonnées. Le worker diffuse chaque modification apportée à cette couche (causée par l’exécution des thunks locaux) à tous les autres workers du cluster, ainsi qu’au planificateur. Ceci est fait sur la base d’un «meilleur effort» : si un message de diffusion n’atteint pas sa destination, il est abandonné¹ sans nouvelle tentative.

Chaque seconde, la couche de métadonnées est sauvegardée sur disque, de manière incrémentale. En cas de crash ou de redémarrage, le worker prendra une seconde ou deux pour recharger l’intégralité de la couche à partir du disque afin de se rappeler ce qu’il faisait.

Garder de grandes bases de données en mémoire

Comme mentionné ci-dessus, la couche de métadonnées peut contenir un million d’entrées. Chaque DAG individuel peut contenir des centaines de milliers de nœuds. Tous ceux-ci ont une longue durée de vie — de quelques minutes à plusieurs heures. Garder en mémoire des millions d’objets de longue durée est très éprouvant pour le ramasse-miettes .NET.

La collecte des déchets en .NET est un sujet complexe (bien qu’il existe une excellente série par Konrad Kokosa pour explorer les détails de bas niveau), mais le problème global résulte d’une combinaison de trois faits :

• Le coût en performance d’une passe de collecte des déchets est proportionnel au nombre d’objets vivants dans la zone de mémoire en cours de collecte. Traiter des millions d’objets, souvent avec des milliards de références à suivre entre eux, prendra plusieurs secondes au ramasse-miettes.
• Pour éviter ce coût, le ramasse-miettes .NET travaille avec des zones de mémoire distinctes, appelées générations, en fonction de l’âge des objets qu’elles contiennent. La plus jeune génération, Gen0, subit des collectes des déchets fréquemment mais ne contient que les objets alloués depuis la dernière passe (donc, très peu). La génération la plus ancienne, Gen2, n’est collectée que si Gen1 et Gen0 ont été collectées sans réussir à libérer suffisamment de mémoire. Cela sera assez rare tant que la plupart des allocations d’objets sont petites et de courte durée.
• Cependant, une opération normale de thunk implique de grands tableaux de valeurs, qui sont alloués sur le Large Object Heap, une zone distincte de Gen0, Gen1 et Gen2. Lorsque le Large Object Heap manque d’espace, une collecte complète des déchets est effectuée, qui collecte également Gen2.

Et Gen2 est l’endroit où se situent les millions d’objets issus des DAGs et de la couche de métadonnées.

Pour éviter cela, nous avons conçu à la fois les DAGs et la couche de métadonnées pour n’utiliser que très peu d’objets.

Chaque DAG se compose de seulement deux allocations — un tableau de nœuds et un tableau d’arêtes, tous deux étant des types valeur non managés, de sorte que le GC n’a même pas besoin de parcourir leur contenu pour suivre les éventuelles références qu’ils pourraient contenir. Lorsqu’un thunk doit être exécuté, il est désérialisé à partir de la représentation binaire du DAG², qui est présente dans la couche de métadonnées.

La couche de métadonnées contient des données de longueur variable, elle est donc construite en découpant des segments d’un grand byte[], en utilisant ref struct et MemoryMarshal.Cast pour manipuler les données sans les copier.

Espace de travail temporaire

Un cluster dispose de 512GiB à 1.5TiB de RAM, et entre 15.36TB et 46.08TB de stockage NVMe. La majeure partie de cet espace est dédiée au stockage des résultats intermédiaires de l’évaluation des thunks.

La RAM est un bien précieux : elle ne représente que 3 % de l’espace de stockage disponible, mais est entre 100× et 1000× plus rapide en lecture et en écriture. Il y a un avantage significatif à s’assurer que les données sur le point d’être lues par un thunk sont déjà présentes en mémoire (ou n’ont jamais quitté la mémoire dès le départ).

De plus, il est pratiquement impossible d’utiliser 100 % de la RAM disponible en .NET — le système d’exploitation a des besoins en mémoire variables et n’a aucun moyen fiable de communiquer au processus .NET qu’il doit libérer de la mémoire, ce qui conduit le processus à être oom-killed (out-of-memory).

Envision résout ce problème en déléguant la gestion des transferts de RAM vers NVMe au système d’exploitation. Nous avons open-sourcé ce code sous le nom de Lokad.ScratchSpace. Cette bibliothèque mappe en mémoire tout l’espace de stockage disponible sur les disques NVMe, et l’expose en tant que blob store que l’application peut utiliser pour :

1. écrire des blocs de données (jusqu’à 2GB chacun) dans l’espace de travail temporaire, soit directement, soit en sérialisant à partir d’un objet managé. Cette opération renvoie un identifiant de bloc.
2. lire des blocs de données à l’aide de leurs identifiants. Cette opération fixe le bloc et l’expose à l’application sous forme d’un ReadOnlySpan<byte>, que l’application devra ensuite copier (ou désérialiser) dans la mémoire managée.

Une fois que l’espace de travail temporaire est plein, les blocs les plus anciens sont supprimés pour faire place à de nouvelles données. Cela signifie qu’il est possible qu’une opération de lecture échoue, si l’identifiant pointe vers un bloc qui a été supprimé, mais cela est une occurrence rare lors de l’exécution d’un script Envision — rarement une seule exécution produit-elle des dizaines de téraoctets. En revanche, cela peut empêcher une nouvelle exécution de réutiliser les résultats d’une exécution précédente.

La clé de l’utilisation d’un espace de travail temporaire mappé en mémoire est que la RAM disponible est répartie entre trois types de pages³ : la mémoire qui appartient aux processus (comme le processus .NET d’Envision), la mémoire qui est une copie exacte, octet par octet, d’une portion d’un fichier sur disque, et la mémoire destinée à être écrite sur un fichier sur disque.

La mémoire qui est une copie d’un fichier sur disque peut, à tout moment, être libérée par le système d’exploitation, et utilisée à d’autres fins — être attribuée à un processus pour son propre usage, ou devenir une copie d’une autre portion d’un fichier sur disque. Bien que ce ne soit pas instantané, ces pages agissent comme un tampon mémoire qui peut être rapidement réassigné à un autre usage. Et tant qu’elles ne sont pas réassignées, le système d’exploitation sait qu’elles contiennent une copie d’une région spécifique de mémoire persistante, et toute demande de lecture pour cette région sera redirigée vers la page existante, évitant ainsi toute lecture depuis le disque.

La mémoire destinée à être écrite sur disque sera éventuellement effectivement écrite et deviendra une copie de la région où elle a été écrite. Cette conversion est limitée par la vitesse d’écriture des disques NVMe (de l’ordre de 1GB/s).

La mémoire qui est attribuée au processus ne peut être reconvertie aux deux autres types sans être libérée par le processus (ce que le ramasse-miettes .NET fera parfois, après qu’une collecte a libéré une grande quantité de mémoire). Toute mémoire allouée via .NET, y compris tous les objets managés et tout ce qui est géré par le GC, doit appartenir à ce type de mémoire.

Dans un worker typique, 25 % de la mémoire est directement attribuée au processus .NET, 70 % est une copie en lecture seule de régions de fichiers, et 5 % est en cours d’écriture sur disque.

Couche d’atomes

Le principe général est que chaque thunk écrit sa sortie dans l’espace de travail temporaire sous forme d’un ou plusieurs atomes, puis stocke les identifiants de ces atomes dans la couche de métadonnées. Les thunks suivants chargent ensuite ces identifiants depuis la couche de métadonnées et les utilisent pour interroger l’espace de travail temporaire afin d’obtenir les atomes dont ils ont besoin.

Le nom «Atome» a été choisi car il est impossible de lire seulement une partie d’un atome : ils ne peuvent être récupérés qu’en totalité. Si une structure de données doit permettre de demander seulement une partie de son contenu, nous la sauvegardons alors sous forme de plusieurs atomes, qui peuvent être récupérés indépendamment.

Certains atomes sont compressés ; par exemple, la plupart des vecteurs de booléens ne sont pas représentés sous forme de bool[], qui consomme un octet par élément, mais sont plutôt compactés à 1 bit par valeur, puis compressés pour éliminer les longues séquences de valeurs identiques.

Il est possible que des atomes disparaissent, bien que cela soit rare. Les deux principales situations où cela peut se produire sont lorsque la couche de métadonnées conserve un résultat d’une exécution précédente, mais que l’atome correspondant a été évincé de l’espace de travail temporaire entre-temps, et lorsque l’atome a été stocké sur un worker différent qui ne répond plus aux requêtes. Moins fréquemment, une erreur de somme de contrôle révèle que les données stockées ne sont plus valides et doivent être écartées.

Lorsqu’un atome disparaît, le thunk qui l’avait demandé est interrompu, et passe en mode de récupération :

1. Le système vérifie la présence (mais pas les sommes de contrôle) de tous les autres atomes référencés par les entrées du thunk. Cela s’explique par le fait que les atomes sont susceptibles d’être générés au même moment et sur le même worker, et la disparition d’un atome est corrélée à la disparition d’autres atomes du même temps et lieu.
2. Le système passe en revue la couche de métadonnées à la recherche de références aux atomes découverts comme manquants lors de l’étape précédente. Cela amènera certains thunks à passer de « exécuté » à « pas encore exécuté » parce que leur résultat a été écarté. Le noyau le détectera alors, et les replanifiera.

Les thunks réexécutés produiront alors à nouveau l’atome, et l’exécution pourra reprendre.

Tableaux d’atomes

Un aspect particulier de la couche d’atomes est la manière dont les mélanges sont effectués — une première couche de $M$ thunks produit chacun plusieurs millions de lignes de données, puis une deuxième couche de $N$ thunks lit la sortie de la couche précédente pour effectuer une autre opération (généralement, une forme de réduction), mais chaque ligne de la première couche n’est lue que par un seul thunk de la deuxième couche.

Il serait très inefficace que chaque thunk de la deuxième couche lise toutes les données de la première couche (chaque ligne serait lue $N$ fois, dont $N-1$ seraient superflues), mais c’est exactement ce qui se passerait si chaque thunk de la première couche produisait exactement un atome.

D’autre part, si chaque thunk dans la première couche produit un atome pour chaque thunk dans la seconde couche, l’opération de shuffle impliquera $M\cdot N$ atomes au total—un million d’atomes pour $M = N = 1000$. Bien que la surcharge sur les atomes ne soit pas excessive, en ajoutant un identifiant d’atome, un identifiant de tenant, le type de données de l’atome, la taille et un peu de tenue de registres, cela peut encore atteindre quelques centaines d’octets par atome. Bien que 100MB puisse sembler un petit prix à payer pour déplacer 4GB de données réelles, ces données réelles résident dans la couche d’atomes (qui est conçue pour de grandes données), tandis que 100MB représente une part conséquente du budget total de 1.5GB de la couche de métadonnées.

Pour contourner ce problème, Envision prend en charge les tableaux d’atomes:

• Tous les atomes d’un tableau d’atomes sont écrits en une fois, et sont conservés ensemble à la fois en mémoire et sur le disque.
• Étant donné l’identifiant du tableau d’atomes, il est facile de déduire l’identifiant du ième atome dans le tableau.

Grâce à cela, un tableau d’atomes présente la même surcharge qu’un seul atome. Dans un shuffle, les thunks de la première couche produiraient $M$ tableaux contenant chacun $N$ atomes. Les thunks de la seconde couche demanderaient chacun $M$ atomes, un de chaque tableau, à la position correspondant au rang de ce thunk dans le shuffle.

Pour conclure, quelques statistiques de production ! En une heure, un ouvrier typique exécutera 150 000 thunks et écrira 200 000 atomes (les tableaux d’atomes ne sont comptés qu’une seule fois) représentant 750GiB de données intermédiaires.

Dans le prochain et dernier article de cette série, nous discuterons des couches qui permettent l’exécution distribuée.

Petit coup de pub : nous recrutons ingénieurs logiciels. Le télétravail est possible.