Envision VM (Teil 3), Atome und Datenspeicherung

Victor Nicollet

Dieser Artikel ist der dritte Teil einer vierteiligen Serie über die inneren Abläufe 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 andermal), also nehmen wir einfach an, dass das Skript irgendwie in den Bytecode umgewandelt wurde, den die Envision Virtual Machine als Eingabe nimmt.

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

• Eine Milliarde boolesche Werte (ein Bit pro Wert) benötigen 125MB.
• Eine Milliarde Gleitkommazahlen (32-Bit-Präzision) benötigen 4GB.
• Eine Milliarde minimale Verkaufszeilen (Datum, Ort, EAN-13, Menge) benötigen zwischen 14GB und 33GB (oder mehr!) je nachdem, wie die Werte kodiert sind.

Dies stellt zwei Herausforderungen dar: Wie man diese Daten vom Moment ihrer Erstellung bis zu ihrer Verwendung sichert (ein Teil der Antwort lautet: auf NVMe-Laufwerken, die über mehrere Maschinen verteilt sind), und wie man die Menge der Daten minimiert, die durch Kanäle fließen, die langsamer als RAM sind (Netzwerk und dauerhafter Speicher).

Metadaten-Schicht

Ein Teil der Lösung besteht darin, zwei separate Datenschichten zu haben, wobei Daten basierend auf ihrer Natur in eine der beiden Schichten geschoben werden. Die Metadaten-Schicht enthält Informationen über die eigentlichen Daten und über die Skripte, die ausgeführt werden:

• 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 vom Thunk erzeugten Fehlermeldungen in dieser Schicht gespeichert.
• Wenn ein Thunk einen neuen Thunk (und den DAG seiner Eltern) zurückgegeben hat, wird der serialisierte DAG in dieser Schicht gespeichert.
• Ein Thunk kann Checkpoints in der Metadaten-Schicht speichern (in der Regel bestehend aus einem Block der Datenkennung); wenn ein Thunk unterbrochen wird, bevor er abgeschlossen wurde, kann er dann seinen Checkpoint aus der Metadaten-Schicht laden und die Arbeit von dieser Position aus fortsetzen.

Mit anderen Worten, die Metadaten-Schicht kann als ein Wörterbuch angesehen werden, das Thunks auf Ergebnisse abbildet, wobei die genaue Natur des Ergebnisses davon abhängt, was der Thunk tatsächlich zurückgegeben hat.

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

Es gibt zwei Beschränkungen für Werte, die in der Metadaten-Schicht gespeichert werden: Ein Eintrag darf 10MB nicht überschreiten (sodass auch ein serialisierter DAG diese Menge nicht überschreiten darf!), und der gesamte Speicherplatz für die Metadaten-Schicht beträgt 1,5GB. Üblicherweise gibt es in dieser Schicht etwa eine Million Werte, bei einer durchschnittlichen Eintragsgröße von 1,5KB.

Die Metadaten-Schicht lebt 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 es ein mit diesem Thunk assoziiertes Ergebnis in der Metadaten-Schicht gibt — wobei dies nicht garantiert, dass die vom Ergebnis referenzierten Daten verfügbar sind.

Jeder Worker im Cluster behält eine eigene Kopie der Metadaten-Schicht. 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 einer «Best-effort»-Basis: Wenn eine Broadcast-Nachricht ihr Ziel nicht erreicht, wird sie ohne erneuten Versuch verworfen¹.

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

Große Datenbanken im Speicher halten

Wie oben erwähnt, kann die Metadaten-Schicht eine Million Einträge enthalten. Jeder einzelne DAG kann Hunderttausende von Knoten enthalten. All diese haben lange Lebensdauern — von Minuten bis zu Stunden. Millionen von langlebigen Objekten im Speicher zu halten, belastet den Garbage Collector von .NET erheblich.

Die Müllabfuhr (Garbage Collection) in .NET ist ein komplexes Thema (obwohl es eine ausgezeichnete Serie von Konrad Kokosa gibt, die sich in die Details vertieft), aber das Gesamtproblem ist eine Kombination aus drei Fakten:

• Die Leistungskosten eines Garbage-Collection-Durchlaufs sind proportional zur Anzahl der lebenden Objekte im Speicherbereich, der gesammelt wird. Das Verarbeiten von Millionen von Objekten, oft mit Milliarden von Referenzen, die zwischen ihnen verfolgt werden müssen, wird den Garbage Collector mehrere Sekunden in Anspruch nehmen.
• Um diese Kosten zu vermeiden, arbeitet der .NET Garbage Collector mit getrennten Speicherbereichen, den sogenannten Generationen, die vom Alter der darin befindlichen Objekte abhängen. Die jüngste Generation, Gen0, wird häufig gesammelt, enthält jedoch nur Objekte, die seit dem letzten Durchlauf zugewiesen wurden (also nur wenige). Die älteste Generation, Gen2, wird nur gesammelt, wenn sowohl Gen1 als auch Gen0 gesammelt wurden, aber nicht genügend freier Speicher zur Verfügung stand. Dies wird ziemlich selten vorkommen, solange die meisten Objektallokationen klein und kurzlebig sind.
• Allerdings beinhaltet ein normaler Thunk-Vorgang große Arrays von Werten, die auf dem Large Object Heap zugeordnet 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, die auch Gen2 sammelt.

Und Gen2 ist der Bereich, in dem sich die Millionen von Objekten aus DAGs und der Metadaten-Schicht befinden.

Um dies zu vermeiden, haben wir sowohl die DAGs als auch die Metadaten-Schicht so gebaut, dass sie nur sehr wenige Objekte verwenden.

Jeder DAG besteht aus nur zwei Allokationen — einem Array von Knoten und einem Array von Kanten, die beide unmanaged Wertetypen sind, sodass der GC nicht einmal deren Inhalte durchlaufen muss, um etwaige Referenzen zu verfolgen. Wenn ein Thunk ausgeführt werden soll, wird er aus der binären Darstellung des DAGs² deserialisiert, die in der Metadaten-Schicht vorhanden ist.

Die Metadaten-Schicht hat Inhalte variabler Länge, weshalb sie aufgebaut wird, indem Blöcke aus einem großen byte[] herausgeschnitten werden, wobei ref struct und MemoryMarshal.Cast verwendet werden, um die Daten ohne Kopieren zu manipulieren.

Zwischenspeicher

Ein Cluster verfügt über zwischen 512GiB und 1.5TiB RAM und zwischen 15.36TB und 46.08TB NVMe-Speicher. Der Großteil dieses Speicherplatzes ist der Speicherung der Zwischenergebnisse von Thunk-Berechnungen gewidmet.

RAM ist wertvoller Besitz: Er repräsentiert nur 3% des verfügbaren Speicherplatzes, ist jedoch zwischen 100× und 1000× schneller im 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 niemals den Speicher verlassen haben).

Zudem ist es in .NET nahezu unmöglich, 100% des verfügbaren RAM zu nutzen — das Betriebssystem hat variable Speicherbedürfnisse und es gibt keinen verlässlichen Mechanismus, dem .NET-Prozess mitzuteilen, dass er etwas Speicher freigeben soll, was dazu führt, dass der Prozess aufgrund von OOM (Out-of-Memory) beendet wird.

Envision löst dieses Problem, indem das Management der RAM-zu-NVMe Transfers an das Betriebssystem delegiert wird. Wir haben diesen Code als Lokad.ScratchSpace Open Source gestellt. Diese Bibliothek bildet den gesamten verfügbaren Speicherplatz der NVMe-Laufwerke als Memory-Mapped View ab und stellt ihn als Blob-Speicher zur Verfügung, den die Anwendung verwenden kann, um:

1. Datenblöcke (bis zu je 2GB) in den Scratch Space zu schreiben, entweder direkt oder durch Serialisierung aus einem verwalteten Objekt. Dieser Vorgang gibt eine Blockkennung zurück.
2. Datenblöcke anhand ihrer Kennungen zu lesen. Dieser Vorgang fixiert den Block und stellt ihn der Anwendung als ReadOnlySpan<byte> zur Verfügung, welchen die Anwendung dann kopieren (oder deserialisieren) sollte, um ihn in den verwalteten Speicher zu übernehmen.

Sobald der Scratch Space voll ist, werden die ältesten Blöcke verworfen, um Platz für neue Daten zu schaffen. Das bedeutet, dass es möglich ist, dass ein Lesevorgang fehlschlägt, wenn die Kennung auf einen Block verweist, der inzwischen entfernt wurde, was während der Ausführung eines Envision-Skripts selten vorkommt — selten produziert eine einzelne Ausführung Dutzende von Terabytes. Andererseits könnte dies verhindern, dass eine neue Ausführung die Ergebnisse einer vorherigen wiederverwendet.

Der Schlüssel zur Nutzung eines Memory-Mapped Scratch Space besteht darin, dass der verfügbare RAM auf drei Arten von Seiten³ verteilt ist: Speicher, der Prozessen zugeordnet ist (wie dem .NET-Prozess von Envision), Speicher, der eine exakte Byte-für-Byte-Kopie eines Teils einer Datei auf der Festplatte ist, und Speicher, der dafür vorgesehen ist, in eine Datei auf die Festplatte geschrieben zu werden.

Speicher, der eine Kopie einer Datei auf der Festplatte darstellt, kann vom Betriebssystem jederzeit freigegeben und für einen anderen Zweck verwendet werden — sei es, einem Prozess zur eigenen Verwendung zugewiesen zu werden oder um eine Kopie eines anderen Teils einer Datei auf der Festplatte zu werden. Auch wenn dies nicht augenblicklich geschieht, wirken diese Seiten als ein Speicherpuffer, der schnell einer anderen Verwendung zugewiesen werden kann. Und solange sie nicht neu zugewiesen werden, weiß das Betriebssystem, dass sie eine Kopie eines bestimmten Bereichs des dauerhaften Speichers enthalten, sodass alle Leseanfragen für diesen Bereich stattdessen an die bereits vorhandene Seite umgeleitet werden, wodurch kein Festplattenzugriff nötig ist.

Speicher, der dafür vorgesehen ist, auf die Festplatte geschrieben zu werden, wird schließlich tatsächlich weggeschrieben und zu einer Kopie des Bereichs, in den er geschrieben wurde. Diese Umwandlung wird durch die Schreibgeschwindigkeit der NVMe-Laufwerke begrenzt (in der Größenordnung von 1GB/s).

Speicher, der einem Prozess zugewiesen ist, kann nicht wieder in die beiden anderen Typen umgewandelt werden, ohne dass er vom Prozess freigegeben wird (was der .NET GC manchmal tut, nachdem eine Collection eine große Menge Speicher freigegeben hat). Jeglicher Speicher, der über .NET allokiert wird, einschließlich aller verwalteten Objekte und allem, was der GC überwacht, muss zu diesem Speichertyp gehören.

In einem typischen Worker sind 25% des Speichers direkt dem .NET-Prozess zugewiesen, 70% sind eine schreibgeschützte Kopie von Dateibereichen, und 5% befinden sich im Schreibvorgang.

Atom-Schicht

Das allgemeine Prinzip ist, dass jeder Thunk seine Ausgabe als einen oder mehrere Atome in den Scratch Space schreibt und dann die Kennungen dieser Atome in der Metadaten-Schicht speichert. Nachfolgende Thunks laden diese Kennungen aus der Metadaten-Schicht und verwenden sie, um im Scratch Space nach den benötigten Atomen zu suchen.

Der Name „Atom“ wurde gewählt, weil es nicht möglich ist, nur einen Teil eines Atoms zu lesen: Sie können nur vollständig abgerufen werden. Muss eine Datenstruktur das Abrufen nur eines Teils ihres Inhalts unterstützen, speichern wir sie stattdessen als mehrere Atome, die dann unabhängig voneinander abgerufen werden können.

Einige Atome sind komprimiert; so werden beispielsweise die meisten booleschen Vektoren nicht als bool[] dargestellt, welches ein Byte pro Element benötigt, sondern stattdessen auf 1 Bit pro Wert verdichtet und anschließend komprimiert, um lange Sequenzen identischer Werte zu eliminieren.

Es ist möglich, dass Atome verschwinden, wenngleich dies selten vorkommt. Die beiden Hauptszenarien, in denen dies geschieht, sind, wenn die Metadaten-Schicht ein Ergebnis einer früheren Ausführung speichert, dessen zugehöriges Atom aber zwischenzeitlich aus dem Scratch Space entfernt wurde, und wenn das Atom auf einem anderen Worker gespeichert wurde, der nicht mehr auf Anfragen reagiert. Seltener ergibt ein Prüfsummenfehler, 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 versetzt in den Wiederherstellungsmodus:

1. Das System überprüft das Vorhandensein (aber nicht die Prüfsummen) aller anderen Atome, die in den Eingaben des Thunks referenziert werden. Dies liegt daran, dass Atome wahrscheinlich gleichzeitig und auf demselben Worker generiert werden, und das Verschwinden eines Atoms mit dem Verschwinden anderer Atome aus demselben Zeitraum und Ort korreliert.
2. Das System durchkämmt die Metadaten-Schicht nach Referenzen auf eines der Atome, die im vorherigen Schritt als fehlend identifiziert wurden. Dies führt dazu, dass einige Thunks von „ausgeführt“ zu „noch nicht ausgeführt“ zurückgesetzt werden, weil ihr Ergebnis verworfen wurde. Der Kernel wird dies erkennen und sie erneut einplanen.

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

Atom-Arrays

Ein besonderer Aspekt der Atom-Schicht ist die Art und Weise, wie Mischvorgänge durchgeführt werden — eine erste Schicht von $M$ Thunks erzeugt jeweils mehrere Millionen Datenzeilen, wonach eine zweite Schicht von $N$ Thunks die Ausgabe der vorherigen Schicht einliest, um eine weitere Operation durchzuführen (in der Regel eine Art Reduktion), wobei jede einzelne Zeile der ersten Schicht nur von einem Thunk der zweiten Schicht gelesen wird.

Es wäre sehr verschwenderisch, wenn jeder Thunk in der zweiten Schicht alle Daten der ersten Schicht einlesen würde (jede Zeile würde $N$-mal gelesen werden, von denen $N-1$ überflüssig wären), aber genau das würde geschehen, wenn jeder Thunk der ersten Schicht genau ein Atom produziert.

Andererseits, wenn jeder Thunk in der ersten Ebene für jeden Thunk in der zweiten Ebene ein Atom erzeugt, wird der Shuffle-Vorgang insgesamt $M\cdot N$ Atome beinhalten—eine Million Atome für $M = N = 1000$. Obwohl der Overhead bei den Atomen nicht übermäßig ist – wenn man einen Atom-Identifikator, Mandanten-Identifikator, Atom-Datentyp, Größe und ein wenig Buchhaltung zusammenrechnet – kann er dennoch ein paar Hundert Bytes pro Atom erreichen. Während 100MB wie ein geringer Preis erscheinen mag, um 4GB an tatsächlichen Daten zu verschieben, befinden sich diese tatsächlichen Daten in der Atom-Ebene (die für große Daten ausgelegt ist), während 100MB einen beträchtlichen Anteil des Gesamtbudgets von 1,5GB der Metadaten-Ebene ausmachen.

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

• Alle Atome in einem Atom-Array werden gleichzeitig geschrieben und verbleiben sowohl im Speicher als auch auf der Festplatte zusammen.
• Ist der Identifikator des Atom-Arrays bekannt, so lässt sich der Identifikator des i-ten Atoms im Array leicht ableiten.

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

Abschließend einige Produktionsstatistiken! In einer Stunde führt ein typischer Worker 150 000 Thunks aus und schreibt 200 000 Atome (Atom-Arrays werden nur einmal gezählt), was 750GiB an Zwischendaten entspricht.

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

Schamloser Hinweis: Wir stellen Softwareingenieure ein. Remote-Arbeit ist möglich.