Envision VM (partie 3), Atomes et Stockage des Données

Victor Nicollet

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

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

• Un milliard de booléens (un bit par valeur) occupent 125 Mo.
• Un milliard de nombres à virgule flottante (précision de 32 bits) occupent 4 Go.
• Un milliard de lignes de ventes minimales (date, emplacement, EAN-13, quantité) occupent entre 14 Go et 33 Go (ou plus !) en fonction de la façon dont les valeurs sont encodées.

Cela crée deux défis : comment préserver ces données depuis leur création jusqu’à leur utilisation (une partie de la réponse : sur des disques NVMe répartis sur plusieurs machines), et comment minimiser la quantité de données qui passe 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, les données étant poussées dans l’une des deux couches en fonction de leur nature. La couche de métadonnées contient des informations sur les données réelles et sur les scripts en cours d’exécution :

• Lorsqu’un thunk a renvoyé des données avec succès, 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 enregistrer 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 ensuite charger son point de contrôle depuis la couche de métadonnées et reprendre le travail à partir de cette position.

En d’autres termes, la couche de métadonnées peut être considérée comme un dictionnaire qui fait correspondre des thunks à des résultats, où la nature exacte du résultat dépend de ce que le thunk a réellement renvoyé.

La couche de métadonnées peut également contenir des informations supplémentaires sur 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 y a deux limites sur les valeurs stockées dans la couche de métadonnées : une entrée ne peut pas dépasser 10 Mo (donc un DAG sérialisé n’est pas autorisé à dépasser cette quantité non plus !), et l’espace de stockage total pour la couche de métadonnées est de 1,5 Go. En général, il y a environ un million de valeurs dans cette couche, pour une taille moyenne d’entrée de 1,5 Ko.

La couche de métadonnées réside toujours en RAM pour garantir un accès rapide. Elle agit comme la source de vérité pour l’exécution des thunks : un thunk a été exécuté si, et seulement si, il existe un résultat associé à ce thunk 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 du cluster conserve sa propre copie de la couche de métadonnées. Le worker diffuse chaque changement de cette couche (causé par l’exécution des thunks locaux) à tous les autres workers du cluster, ainsi qu’au planificateur. Cela se fait sur la base du “meilleur effort” : si un message de diffusion n’atteint pas sa destination, il est abandonné¹ sans nouvelle tentative.

Toutes les secondes, la couche de métadonnées est persistée de manière incrémentielle sur le disque. En cas de panne ou de redémarrage, le worker prendra une seconde ou deux pour recharger l’intégralité de la couche depuis le disque afin de se souvenir de ce qu’il faisait.

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

Comme mentionné précédemment, 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 ces objets ont une longue durée de vie, de quelques minutes à quelques heures. Garder des millions d’objets à longue durée de vie en mémoire est assez difficile pour le garbage collector de .NET.

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

• Le coût de performance d’un passage de collecte des déchets est proportionnel au nombre d’objets vivants dans la zone de mémoire en cours de collecte des déchets. Traiter des millions d’objets, souvent avec des milliards de références à suivre entre eux, prendra plusieurs secondes au garbage collector pour les traiter.
• Pour éviter de payer ce coût, le garbage collector de .NET travaille avec des zones de mémoire séparées, appelées générations, en fonction de l’âge des objets qu’elles contiennent. La génération la plus jeune, Gen0, subit fréquemment une collecte des déchets mais ne contient que les objets alloués depuis le dernier passage (donc seulement quelques-uns). La génération la plus ancienne, Gen2, n’est collectée que si Gen1 et Gen0 ont été collectées mais n’ont pas réussi à libérer suffisamment de mémoire libre. Cela sera assez rare tant que la plupart des allocations d’objets sont petites et de courte durée.
• Cependant, une opération de thunk normale implique de grandes matrices de valeurs, qui sont allouées sur le Large Object Heap, une zone distincte de Gen0, Gen1 et Gen2. Lorsque le Large Object Heap est saturé, une collecte complète des déchets est effectuée, qui collecte également Gen2.

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

Pour éviter cela, nous avons construit à la fois les DAG 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 de valeur non géré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 peuvent contenir. Lorsqu’un thunk doit être exécuté, il est désérialisé à partir de la représentation binaire du DAG², qui est présent dans la couche de métadonnées.

La couche de métadonnées a un contenu de longueur variable, elle est donc construite en découpant des morceaux 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 512 Gio à 1,5 Tio de RAM et de 15,36 à 46,08 Tio de stockage NVMe. La plupart de cet espace est dédié au stockage des résultats intermédiaires de l’évaluation des thunks.

La RAM est un espace précieux : elle représente seulement 3 % de l’espace de stockage disponible, mais elle est entre 100 et 1000 fois plus rapide en lecture et en écriture. Il est très avantageux de s’assurer que les données qui vont être lues par un thunk sont déjà présentes en mémoire (ou n’ont jamais quitté la mémoire en premier lieu).

De plus, il est presque impossible d’utiliser 100 % de la RAM disponible dans .NET - le système d’exploitation a des besoins variables en mémoire et n’a aucun moyen fiable de communiquer au processus .NET qu’il doit libérer de la mémoire, ce qui entraîne la suppression du processus par manque de mémoire (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 rendu ce code open source sous le nom de Lokad.ScratchSpace. Cette bibliothèque mappes en mémoire tout l’espace de stockage disponible sur les disques NVMe et le expose en tant que magasin de blocs que l’application peut utiliser pour :

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

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

La clé pour utiliser 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 partie d’un fichier sur disque, et la mémoire qui est destinée à être écrite dans un fichier sur disque.

La mémoire qui est une copie d’un fichier sur disque peut, à n’importe quel moment, être libérée par le système d’exploitation et utilisée à d’autres fins - pour être donnée à un processus pour son propre usage, ou pour devenir une copie d’une autre partie d’un fichier sur disque. Bien que cela ne soit pas instantané, ces pages agissent comme un tampon de mémoire qui peut être rapidement réaffecté à une autre utilisation. Et tant qu’elles ne sont pas réaffectées, le système d’exploitation sait qu’elles contiennent une copie d’une région spécifique de la mémoire persistante, de sorte que toutes les demandes de lecture pour cette région seront redirigées vers la page existante, ne nécessitant ainsi aucun chargement à partir du disque.

La mémoire destinée à être écrite sur disque sera finalement é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 1 Go/s).

La mémoire assignée au processus ne peut pas être convertie en ces deux autres types sans être libérée par le processus (ce que le GC .NET fait parfois, après qu’une collection a libéré une grande quantité de mémoire). Toute la mémoire allouée via .NET, y compris tous les objets gérés et tout ce que le GC supervise, doit appartenir à ce type de mémoire.

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

Couche Atom

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

Le nom «Atom» a été choisi car il n’est pas possible de lire seulement une partie d’un atome : ils ne peuvent être récupérés que dans leur intégralité. Si une structure de données doit prendre en charge la demande d’une partie seulement de son contenu, nous la sauvegardons plutôt sous forme de plusieurs atomes, qui peuvent ensuite être récupérés indépendamment.

Certains atomes sont compressés ; par exemple, la plupart des vecteurs booléens ne sont pas représentés sous forme de bool[], ce qui consomme un octet par élément, mais sont plutôt réduits à 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 se souvient d’un résultat d’une exécution précédente, mais que l’atome correspondant a été évacué de l’espace de stockage temporaire entre-temps, et lorsque l’atome était stocké sur un autre travailleur qui ne répond plus aux demandes. 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 supprimées.

Lorsqu’un atome disparaît, le thunk qui l’a 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 est dû au fait que les atomes sont susceptibles d’être générés en même temps et sur le même travailleur, et la disparition d’un atome est corrélée à la disparition d’autres atomes provenant de la même période et du même endroit.
2. Le système parcourt la couche de métadonnées à la recherche de références à l’un des atomes découverts comme manquants lors de l’étape précédente. Cela fera passer certains thunks de l’état “exécuté” à “pas encore exécuté” car leur résultat a été supprimé. Le noyau le détectera alors et les planifiera à nouveau.

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 chacune 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 inutiles), mais c’est exactement ce qui se produirait si chaque thunk de la première couche produisait exactement un atome.

D’autre part, si chaque thunk de la première couche produit un atome pour chaque thunk de la deuxième couche, l’opération de mélange impliquera un total de $M\cdot N$ atomes—un million d’atomes pour $M = N = 1000$. Bien que le surcoût des atomes ne soit pas excessif, en ajoutant un identifiant d’atome, un identifiant de locataire, un type de données d’atome, une taille et un peu de gestion, cela peut atteindre quelques centaines d’octets par atome. Bien que 100 Mo puisse sembler un prix modeste à payer pour mélanger environ 4 Go de données réelles, ces données réelles résident dans la couche d’atomes (qui est conçue pour les grandes données), tandis que 100 Mo représentent une part importante du budget total de 1,5 Go 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 même temps 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 a le même surcoût qu’un seul atome. Dans un mélange, les thunks de la première couche produiraient $M$ tableaux de $N$ atomes chacun. Les thunks de la deuxième couche demanderaient chacun $M$ atomes, un de chaque tableau, à la position correspondant au rang de ce thunk dans le mélange.

En conclusion, voici quelques statistiques de production ! En une heure, un travailleur typique exécutera 150 000 thunks et écrira 200 000 atomes (les tableaux d’atomes ne sont comptés qu’une seule fois), ce qui représente 750 Gio 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.

Petite publicité : nous recrutons des ingénieurs logiciels. Le travail à distance est possible.