Envision VM (Teil 3), Atome und Datenspeicherung

Victor Nicollet

Dieser Artikel ist der dritte Teil einer vierteiligen Serie über die Funktionsweise der Envision-Virtual Machine: die Software, die Envision-Skripte ausführt. Siehe Teil 1, Teil 2 und Teil 4. Diese Serie behandelt nicht den Envision-Compiler (vielleicht ein anderes Mal), daher nehmen wir einfach an, dass das Skript irgendwie in den Bytecode umgewandelt wurde, den die Envision-Virtual Machine als Eingabe akzeptiert.

Während der Ausführung lesen Thunks Eingabedaten und schreiben Ausgabedaten, oft in großen Mengen.

• Eine Milliarde Booleans (ein Bit pro Wert) benötigen 125 MB.
• Eine Milliarde Gleitkommazahlen (32-Bit-Genauigkeit) benötigen 4 GB.
• Eine Milliarde minimale Verkaufslinien (Datum, Ort, EAN-13, Menge) benötigen je nach Codierung zwischen 14 GB und 33 GB (oder mehr!).

Dies stellt zwei Herausforderungen dar: Wie können diese Daten von dem Moment an, in dem sie erstellt werden, bis zu dem Moment, in dem sie verwendet werden, aufbewahrt werden (ein Teil der Antwort liegt auf NVMe-Laufwerken, die auf mehrere Maschinen verteilt sind), und wie kann die Menge an Daten minimiert werden, die durch Kanäle fließen, die langsamer als der RAM sind (Netzwerk und persistenter Speicher).

Metadatenschicht

Ein Teil der Lösung besteht darin, zwei separate Datenebenen zu haben, wobei Daten basierend auf ihrer Natur in eine der beiden Ebenen geschoben werden. Die Metadatenschicht enthält Informationen über die tatsächlichen Daten und über die ausgeführten Skripte:

• Wenn ein Thunk erfolgreich Daten zurückgegeben hat, wird die eindeutige Kennung dieser Daten in dieser Schicht gespeichert.
• Wenn ein Thunk fehlgeschlagen ist, werden die von dem Thunk erzeugten Fehlermeldungen in dieser Schicht gespeichert.
• Wenn ein Thunk einen neuen Thunk zurückgegeben hat (und den DAG seiner Eltern), wird der serialisierte DAG in dieser Schicht gespeichert.
• Ein Thunk kann Checkpoints in der Metadatenschicht speichern (normalerweise besteht ein Checkpoint aus der Kennung eines Datenblocks); wenn ein Thunk unterbrochen wird, bevor er abgeschlossen wurde, kann er seinen Checkpoint aus der Metadatenschicht laden und die Arbeit an dieser Position fortsetzen.

Mit anderen Worten kann die Metadatenschicht als ein Wörterbuch angesehen werden, das Thunks den Ergebnissen zuordnet, wobei die genaue Art des Ergebnisses davon abhängt, was der Thunk tatsächlich zurückgegeben hat.

Die Metadatenschicht kann auch zusätzliche Informationen über die Struktur der referenzierten Daten enthalten. Wenn beispielsweise ein Thunk ein Paar von Vektoren zurückgegeben hat, enthält die Metadaten die eindeutige Kennung jedes Vektors. Dies ermöglicht es Verbrauchern, auf einen Vektor zuzugreifen, ohne beide laden zu müssen.

Es gibt zwei Grenzen für Werte, die in der Metadatenschicht gespeichert sind: Ein Eintrag darf 10 MB nicht überschreiten (ein serialisierter DAG darf also auch diese Menge nicht überschreiten!) und der Gesamtspeicherplatz für die Metadatenschicht beträgt 1,5 GB. In der Regel gibt es in dieser Schicht etwa eine Million Werte, was zu einer durchschnittlichen Eintragsgröße von 1,5 KB führt.

Die Metadatenschicht befindet sich immer im RAM, um einen schnellen Zugriff zu gewährleisten. Sie fungiert als Quelle der Wahrheit für die Ausführung von Thunks: Ein Thunk wurde ausgeführt, wenn und nur wenn diesem Thunk ein Ergebnis in der Metadatenschicht zugeordnet ist - obwohl dies nicht garantiert, dass die von diesem Ergebnis referenzierten Daten verfügbar sind.

Jeder Worker im Cluster behält eine eigene Kopie der Metadatenschicht. Der Worker sendet jede Änderung dieser Schicht (verursacht durch die Ausführung lokaler Thunks) an alle anderen Worker im Cluster und auch an den Scheduler. Dies geschieht auf “best effort”-Basis: Wenn eine Broadcast-Nachricht ihr Ziel nicht erreicht, wird sie ohne erneuten Versuch verworfen¹.

Alle paar Sekunden wird die Metadatenschicht inkrementell auf die Festplatte geschrieben. Im Falle eines Absturzes oder Neustarts benötigt der Worker eine Sekunde oder zwei, um die gesamte Schicht von der Festplatte neu zu laden und sich daran zu erinnern, was er getan hat.

Große Datenbanken im Speicher halten

Wie oben erwähnt, kann die Metadatenschicht eine Million Einträge enthalten. Jeder einzelne DAG kann Hunderttausende von Knoten enthalten. All diese haben eine lange Lebensdauer - von Minuten bis Stunden. Millionen von langlebigen Objekten im Speicher zu halten, ist für den .NET Garbage Collector ziemlich schwierig.

Die Garbage Collection in .NET ist ein komplexes Thema (obwohl es eine ausgezeichnete Serie von Konrad Kokosa gibt, um in die Details auf niedriger Ebene einzutauchen), aber das Hauptproblem besteht aus drei Fakten:

• Die Leistungskosten eines Garbage Collection-Durchlaufs sind proportional zur Anzahl der lebenden Objekte im Bereich des zu sammelnden Speichers. Die Verarbeitung von Millionen von Objekten, oft mit Milliarden von Referenzen zwischen ihnen, dauert mehrere Sekunden, um vom Garbage Collector verarbeitet zu werden.
• Um diese Kosten zu vermeiden, arbeitet der .NET Garbage Collector mit separaten Speicherbereichen, die je nach Alter der darin enthaltenen Objekte als Generationen bezeichnet werden. Die jüngste Generation, Gen0, wird häufig einer Garbage Collection unterzogen, enthält jedoch nur Objekte, die seit dem letzten Durchlauf zugewiesen wurden (also nur wenige). Die älteste Generation, Gen2, wird nur dann gesammelt, wenn sowohl Gen1 als auch Gen0 gesammelt wurden, aber nicht genügend freien Speicherplatz ergeben haben. Dies wird recht selten sein, solange die meisten Objektzuweisungen klein und kurzlebig sind.
• Eine normale Thunk-Operation umfasst jedoch große Arrays von Werten, die auf dem Large Object Heap allokiert werden, einem Bereich, der von Gen0, Gen1 und Gen2 getrennt ist. Wenn der Large Object Heap keinen Platz mehr hat, wird eine vollständige Garbage Collection durchgeführt, bei der auch Gen2 gesammelt wird.

Und Gen2 ist der Ort, an dem sich die Millionen von Objekten aus DAGs und der Metadatenschicht befinden.

Um dies zu vermeiden, haben wir sowohl die DAGs als auch die Metadatenschicht so konstruiert, dass sie nur sehr wenige Objekte verwenden.

Jeder DAG besteht nur aus zwei Zuweisungen - einem Array von Knoten und einem Array von Kanten, die beide nicht verwaltete Werttypen sind, sodass der GC nicht einmal den Inhalt durchsuchen muss, um etwaige enthaltene Referenzen zu verfolgen. Wenn ein Thunk zur Ausführung benötigt wird, wird er aus der binären Repräsentation des DAGs deserialisiert², die in der Metadatenschicht vorhanden ist.

Die Metadatenschicht hat Inhalte variabler Länge, daher wird sie durch das Ausschneiden von Chunks aus einem großen byte[] erstellt, wobei ref struct und MemoryMarshal.Cast verwendet werden, um die Daten ohne Kopieren zu manipulieren.

Scratch Space

Ein Cluster verfügt über 512GiB bis 1,5TiB RAM und 15,36TB bis 46,08TB NVMe-Speicher. Der Großteil dieses Speicherplatzes dient zur Speicherung der Zwischenergebnisse der Thunk-Auswertung.

RAM ist wertvoller Speicherplatz: Er repräsentiert nur 3% des verfügbaren Speicherplatzes, ist aber zwischen 100- und 1000-mal schneller beim Lesen und Schreiben. Es gibt einen erheblichen Vorteil darin, sicherzustellen, dass Daten, die von einem Thunk gelesen werden sollen, bereits im Speicher vorhanden sind (oder den Speicherplatz überhaupt nicht verlassen haben).

Darüber hinaus ist es nahezu unmöglich, 100% des verfügbaren RAMs in .NET zu verwenden - das Betriebssystem hat variable Speicherbedürfnisse und es gibt keine zuverlässige Möglichkeit, dem .NET-Prozess mitzuteilen, dass er etwas Speicher freigeben soll. Dies führt dazu, dass der Prozess aufgrund von Speichermangel beendet wird (out-of-memory).

Envision löst dieses Problem, indem es die Verwaltung von RAM-zu-NVMe-Übertragungen dem Betriebssystem überlässt. Wir haben diesen Code als Lokad.ScratchSpace Open Source zur Verfügung gestellt. Diese Bibliothek bildet den gesamten verfügbaren Speicherplatz auf den NVMe-Laufwerken als Blob-Speicher ab, den die Anwendung nutzen kann, um:

1. Datenblöcke (bis zu 2 GB pro Block) direkt oder durch Serialisierung aus einem verwalteten Objekt in den Scratch Space zu schreiben. Diese Operation gibt einen Block-Identifier zurück.
2. Datenblöcke anhand ihrer Identifikatoren lesen. Diese Operation fixiert den Block und stellt ihn der Anwendung als ReadOnlySpan<byte> zur Verfügung, den die Anwendung dann in den verwalteten Speicher kopieren (oder deserialisieren) sollte.

Sobald der Scratch Space voll ist, werden die ältesten Blöcke verworfen, um Platz für neue Daten zu schaffen. Dies bedeutet, dass eine Leseoperation fehlschlagen kann, wenn der Identifier auf einen Block zeigt, der verworfen wurde. Dies ist jedoch ein seltenes Ereignis während der Ausführung eines Envision-Skripts - selten erzeugt eine einzelne Ausführung mehrere Terabyte. Andererseits kann dies verhindern, dass eine neue Ausführung die Ergebnisse einer vorherigen wiederverwendet.

Der Schlüssel zur Verwendung eines memory-mapped Scratch Space besteht darin, dass der verfügbare RAM auf drei Arten von Seiten verteilt ist³: Speicher, der zu Prozessen gehört (wie dem .NET-Prozess von Envision), Speicher, der eine exakte Byte-für-Byte-Kopie eines auf der Festplatte befindlichen Dateiteils ist, und Speicher, der zum Schreiben in eine Datei auf der Festplatte vorgesehen ist.

Speicher, der eine Kopie einer Datei auf der Festplatte ist, kann jederzeit vom Betriebssystem freigegeben und für einen anderen Zweck verwendet werden - entweder um einem Prozess für dessen eigenen Gebrauch zur Verfügung gestellt zu werden oder um eine Kopie eines anderen Teils einer Datei auf der Festplatte zu werden. Obwohl dies nicht sofort geschieht, fungieren diese Seiten als ein Speicherbuffer, der schnell einem anderen Zweck zugewiesen werden kann. Und solange sie nicht erneut zugewiesen werden, weiß das Betriebssystem, dass sie eine Kopie eines bestimmten Bereichs des persistenten Speichers enthalten, und daher werden alle Leseanfragen für diesen Bereich stattdessen zur vorhandenen Seite umgeleitet, was keine Ladevorgänge von der Festplatte erfordert.

Speicher, der zum Schreiben auf die Festplatte vorgesehen ist, wird schließlich geschrieben und wird zu einer Kopie des Bereichs, in dem er geschrieben wurde. Diese Umwandlung ist durch die Schreibgeschwindigkeit der NVMe-Laufwerke begrenzt (in der Größenordnung von 1 GB/s).

Speicher, der dem Prozess zugewiesen ist, kann nicht ohne Freigabe durch den Prozess selbst in die beiden anderen Typen zurückgewandelt werden (was der .NET GC manchmal tut, nachdem eine Sammlung eine große Menge Speicher freigegeben hat). Alle über .NET zugewiesener Speicher, einschließlich aller verwalteten Objekte und allem, was der GC überwacht, muss zu diesem Typ von Speicher gehören.

In einem typischen Worker ist 25% des Speichers direkt dem .NET-Prozess zugewiesen, 70% sind eine schreibgeschützte Kopie von Dateibereichen und 5% werden gerade geschrieben.

Atom Layer

Das allgemeine Prinzip besteht darin, dass jeder Thunk seine Ausgabe als einen oder mehrere Atoms in den Scratch Space schreibt und dann die Identifikatoren dieser Atome in der Metadatenebene speichert. Anschließend laden nachfolgende Thunks diese Identifikatoren aus der Metadatenebene und verwenden sie, um den Scratch Space nach den benötigten Atomen abzufragen.

Der Name «Atom» wurde gewählt, weil es nicht möglich ist, nur einen Teil eines Atoms zu lesen: Sie können nur in ihrer Gesamtheit abgerufen werden. Wenn eine Datenstruktur die Anforderung unterstützen muss, nur einen Teil ihres Inhalts anzufordern, speichern wir sie stattdessen als mehrere Atome, die dann unabhängig voneinander abgerufen werden können.

Einige Atome sind komprimiert; zum Beispiel werden die meisten booleschen Vektoren nicht als bool[] dargestellt, was einen Byte pro Element verbraucht, sondern werden auf 1 Bit pro Wert komprimiert und dann komprimiert, um lange Sequenzen identischer Werte zu eliminieren.

Es ist möglich, dass Atome verschwinden, obwohl dies selten vorkommt. Die beiden Hauptfälle, in denen dies geschehen kann, sind zum einen, wenn die Metadatenebene ein Ergebnis aus einem früheren Lauf speichert, das entsprechende Atom jedoch in der Zwischenzeit aus dem Scratch Space entfernt wurde, und zum anderen, wenn das Atom auf einem anderen Worker gespeichert war, der nicht mehr auf Anfragen antwortet. Seltener führt ein Prüfsummenfehler dazu, dass die gespeicherten Daten nicht mehr gültig sind und verworfen werden müssen.

Wenn ein Atom verschwindet, wird der Thunk, der es angefordert hat, unterbrochen und wechselt in den Wiederherstellungsmodus:

1. Das System überprüft die Anwesenheit (aber nicht die Prüfsummen) aller anderen Atome, auf die der Thunk verweist. Dies liegt daran, dass Atome wahrscheinlich zur gleichen Zeit und auf demselben Worker generiert werden, und das Verschwinden eines Atoms korreliert mit dem Verschwinden anderer Atome aus derselben Zeit und demselben Ort.
2. Das System durchsucht die Metadatenebene nach Verweisen auf eines der während des vorherigen Schritts als fehlend erkannten Atome. Dadurch werden einige Thunks von “ausgeführt” auf “noch nicht ausgeführt” zurückgesetzt, da ihr Ergebnis verworfen wurde. Der Kernel erkennt dies dann und plant sie erneut.

Die erneut ausgeführten Thunks werden dann das Atom erneut erzeugen, und die Ausführung kann fortgesetzt werden.

Atom Arrays

Ein besonderer Aspekt der Atom-Ebene ist die Art und Weise, wie Shuffles durchgeführt werden - eine erste Schicht von $M$ Thunks erzeugt jeweils mehrere Millionen Zeilen Daten, und dann liest eine zweite Schicht von $N$ Thunks die Ausgabe der vorherigen Schicht, um eine weitere Operation durchzuführen (in der Regel eine Form der Reduzierung), aber jede einzelne Zeile aus der ersten Schicht wird nur von einem Thunk aus der zweiten Schicht gelesen.

Es wäre sehr verschwenderisch, wenn jeder Thunk in der zweiten Schicht alle Daten aus der ersten Schicht lesen würde (jede Zeile würde $N$ Mal gelesen werden, von denen $N-1$ unnötig wären), aber genau das würde passieren, wenn jeder Thunk aus der ersten Schicht genau ein Atom erzeugen würde.

Andererseits, wenn jeder Thunk in der ersten Schicht ein Atom für jeden Thunk in der zweiten Schicht erzeugt, wird die Shuffle-Operation insgesamt $M\cdot N$ Atome umfassen - eine Million Atome für $M = N = 1000$. Obwohl der Overhead für Atome nicht übermäßig ist, wenn man einen Atom-Identifier, einen Mandanten-Identifier, den Atom-Datentyp, die Größe und etwas Buchhaltung hinzufügt, kann er immer noch einige hundert Bytes pro Atom erreichen. Während 100 MB vielleicht wie ein kleiner Preis erscheinen, um 4 GB tatsächliche Daten zu verschieben, leben diese tatsächlichen Daten in der Atom-Ebene (die für große Daten ausgelegt ist), während 100 MB einen beträchtlichen Teil des 1,5 GB Gesamtbudgets der Metadatenebene darstellen.

Um dies zu umgehen, unterstützt Envision Atom-Arrays:

• Alle Atome in einem Atom-Array werden gleichzeitig geschrieben und sowohl im Speicher als auch auf der Festplatte zusammengehalten.
• Mit der Kennung des Atom-Arrays ist es einfach, die Kennung des i-ten Atoms im Array abzuleiten.

Dank dessen hat ein Array von Atomen den gleichen Overhead wie ein einzelnes Atom. Bei einem Shuffle würden die Thunks der ersten Schicht $M$ Arrays von jeweils $N$ Atomen erzeugen. Die Thunks der zweiten Schicht würden jeweils $M$ Atome anfordern, eines aus jedem Array, an der Position, die der Rangfolge dieses Thunks im Shuffle entspricht.

Abschließend ein paar Produktionsstatistiken! In einer Stunde wird ein typischer Worker 150.000 Thunks ausführen und 200.000 Atome schreiben (Arrays von Atomen werden nur einmal gezählt), was 750 GiB an Zwischendaten entspricht.

Im nächsten und letzten Artikel dieser Serie werden wir die Ebenen diskutieren, die die verteilte Ausführung ermöglichen.

Unverschämte Werbung: Wir suchen Softwareingenieure. Remote-Arbeit ist möglich.