Envision VM (часть 3), Атомы и хранение данных

Victor Nicollet

Эта статья является третьей частью четырехчастной серии о внутренней работе виртуальной машины Envision: программного обеспечения, которое выполняет скрипты Envision. См. часть 1, часть 2 и часть 4. В этой серии не рассматривается компилятор Envision (возможно, в другой раз), поэтому давайте просто предположим, что скрипт каким-то образом был преобразован в байт-код, который принимает в качестве входных данных виртуальная машина Envision.

Во время выполнения, thunks считывают входные данные и записывают выходные данные, часто в больших объемах.

• Миллиард булевых значений (по одному биту на значение) занимают 125 МБ.
• Миллиард чисел с плавающей запятой (32-битная точность) занимают 4 ГБ.
• Миллиард минимальных строк продаж (дата, местоположение, EAN-13, количество) занимают от 14 ГБ до 33 ГБ (или даже больше!), в зависимости от того, как значения закодированы.

Это создает две проблемы: как сохранить эти данные с момента их создания и до их использования (часть ответа: на NVMe-накопителях, распределенных по нескольким машинам), и как минимизировать количество данных, которые проходят через каналы, медленнее оперативной памяти (сеть и постоянное хранилище).

Слой метаданных

Одна часть решения заключается в наличии двух отдельных слоев данных, в которые данные помещаются в зависимости от их характера. Слой метаданных содержит информацию о фактических данных и о выполняемых скриптах:

• Когда thunk успешно возвращает данные, уникальный идентификатор этих данных сохраняется в этом слое.
• Когда thunk завершается с ошибкой, сообщения об ошибках, созданные thunk, сохраняются в этом слое.
• Когда thunk возвращает новый thunk (и DAG его родителей), сериализованный DAG сохраняется в этом слое.
• Thunk может сохранять контрольные точки в слое метаданных (обычно состоящие из идентификатора блока данных); если thunk прерывается до завершения, он может загрузить свою контрольную точку из слоя метаданных и продолжить работу с этой позиции.

Другими словами, слой метаданных можно рассматривать как словарь, сопоставляющий thunks с результатами, где конкретный характер результата зависит от того, что thunk фактически вернул.

Слой метаданных также может содержать дополнительную информацию о структуре ссылаемых данных. Например, если thunk возвращает пару векторов, то метаданные будут содержать уникальные идентификаторы каждого вектора. Это позволяет потребителям получать доступ к одному вектору, не загружая оба.

Есть два ограничения на значения, хранящиеся в слое метаданных: запись не может превышать 10 МБ (поэтому сериализованный DAG также не может превышать этот объем!), и общий объем хранилища для слоя метаданных составляет 1,5 ГБ. Обычно в этом слое содержится около миллиона значений, средний размер записи составляет 1,5 КБ.

Слой метаданных всегда находится в оперативной памяти, чтобы гарантировать быстрый доступ. Он выступает в качестве источника истины для выполнения thunk: thunk был выполнен, если и только если для этого thunk в слое метаданных есть результат, хотя это не гарантирует доступность данных, на которые ссылается этот результат.

Каждый рабочий в кластере хранит собственную копию слоя метаданных. Рабочий передает каждое изменение этого слоя (вызванное выполнением локальных thunk) всем остальным рабочим в кластере и планировщику. Это делается на основе «наилучших усилий»: если сообщение о передаче не достигает своего назначения, оно отбрасывается¹ без повторной попытки.

Каждую секунду слой метаданных инкрементально сохраняется на диск. В случае сбоя или перезагрузки рабочий затратит одну-две секунды, чтобы загрузить весь слой с диска и вспомнить, что он делал.

Хранение больших баз данных в памяти

Как уже упоминалось выше, слой метаданных может содержать миллион записей. Каждый отдельный DAG может содержать сотни тысяч узлов. Все они имеют длительный срок службы—от нескольких минут до нескольких часов. Хранение миллионов долгоживущих объектов в памяти довольно сложно для сборщика мусора .NET.

Сборка мусора в .NET - это сложная тема (хотя есть отличная серия от Конрада Кокосы, чтобы углубиться в детали низкого уровня), но общая проблема заключается в сочетании трех факторов:

• Стоимость производительности прохода сборки мусора пропорциональна количеству живых объектов в области памяти, подлежащей сборке мусора. Обработка миллионов объектов, часто с миллиардами ссылок между ними, займет несколько секунд для обработки сборщиком мусора.
• Чтобы избежать уплаты этой стоимости, сборщик мусора .NET работает с отдельными областями памяти, называемыми поколениями, в зависимости от возраста объектов внутри них. Самое молодое поколение, Gen0, часто подвергается сборке мусора, но содержит только объекты, выделенные с момента последнего прохода (так что только несколько). Самое старое поколение, Gen2, собирается только в том случае, если и Gen1, и Gen0 были собраны, но не удалось получить достаточно свободной памяти. Это будет довольно редким, пока большинство выделений объектов будут маленькими и краткосрочными.
• Однако обычная операция thunk включает большие массивы значений, которые выделяются в куче больших объектов, области, отличной от Gen0, Gen1 и Gen2. Когда куча больших объектов исчерпывает свое пространство, выполняется полная сборка мусора, которая также собирает Gen2.

Именно в Gen2 находятся миллионы объектов из DAG и слоя метаданных.

Чтобы избежать этого, мы построили как DAG, так и слой метаданных, чтобы использовать только очень немного объектов.

Каждый DAG состоит только из двух выделений - массива узлов и массива ребер, оба из которых являются неуправляемыми типами значений, так что сборщику мусора даже не нужно обходить их содержимое, чтобы следовать любым ссылкам, которые они могут содержать. Когда требуется выполнить thunk, он десериализуется из двоичного представления DAG², которое присутствует в слое метаданных.

Слой метаданных имеет содержимое переменной длины, поэтому он создается путем вырезания фрагментов из большого byte[] с использованием ref struct и MemoryMarshal.Cast для манипулирования данными без их копирования.

Временное пространство

A cluster has between 512GiB and 1.5TiB of RAM, and between 15.36TB and 46.08TB of NVMe storage. Most of this space is dedicated to storing the intermediate results of thunk evaluation.

RAM is valuable real estate: it represents only 3% of available storage space, but is between 100× and 1000× faster to read and write. There is a significant benefit to ensuring that data that is about to be read by a thunk is already present in memory (or has never left memory in the first place).

In addition, it is nearly impossible to use 100% of available RAM in .NET—the operating system has variable memory needs, and has no reliable way of communicating to the .NET process that it should relinquish some memory, resulting in the process being oom-killed (out-of-memory).

Envision resolves this issue by delegating the management of RAM-to-NVMe transfers to the operating system. We have open sourced this code as Lokad.ScratchSpace. This library memory-maps all the storage space available on the NVMe drives, and exposes it as a blob store that the application can use to:

1. write blocks of data (up to 2GB each) to the scratch space, either directly or by serializing from a managed object. This operation returns a block identifier.
2. read blocks of data using their identifiers. This operation pins the block and exposes it to the application as a ReadOnlySpan<byte>, which the application should then copy (or deserialize) to managed memory.

Once the scratch space is full, the oldest blocks are discarded to make space for new data. This means that it’s possible for a read operation to fail, if the identifier points to a block that has been dropped, but this is a rare occurrence during the execution of an Envision script—rarely does a single execution produce tens of terabytes. On the other hand, this may prevent a new execution from reusing the results of a previous one.

The key to using a memory-mapped scratch space is that the available RAM is distributed among three kinds of pages³: memory that belongs to processes (such as Envision’s .NET process), memory that is an exact byte-for-byte copy of an on disk file portion, and memory that is intended to be written to a file on disk.

Memory that is a copy of a file on disk can, at any point in time, be released by the operating system, and used for another purpose—to be given to a process for its own use, or to become a copy of another portion of a file on disk. While not instantaneous, these pages act as a memory buffer that can be quickly re-assigned to another use. And until they are re-assigned, the operating system knows that they contain a copy of a specific region of persistent memory, and so any read requests for that region will be redirected to the existing page instead, thereby requiring no load from disk at all.

Memory that is intended to be written to disk, will eventually be written out and become a copy of the region where it was written. This conversion is limited by the writing speed of the NVMe drives (on the order of 1GB/s).

Memory that is assigned to the process cannot be converted back to the two other types without being released by the process (which the .NET GC will sometimes do, after a collection has released a large amount of memory). All memory allocated through .NET, including all managed objects and everything that the GC oversees, must belong to this type of memory.

In a typical worker, 25% of the memory is assigned to the .NET process directly, 70% is a read-only copy of file regions, and 5% is in the process of being written out.

Слой атомов

Общий принцип заключается в том, что каждый thunk записывает свой вывод во временное пространство в виде одного или нескольких атомов, а затем сохраняет идентификаторы этих атомов в метаданных. Последующие thunk-и затем загружают эти идентификаторы из метаданных и используют их для запроса атомов, которые им нужны.

Имя «Atom» было выбрано, потому что невозможно прочитать только одну часть атома: они могут быть получены только в своей полноте. Если структура данных должна поддерживать запрос только части своего содержимого, мы вместо этого сохраняем ее как несколько атомов, которые затем могут быть получены независимо.

Некоторые атомы сжаты; например, большинство булевых векторов не представлены в виде bool[], который занимает один байт на элемент, а вместо этого сжимаются до 1 бита на значение, а затем сжимаются для устранения длинных последовательностей одинаковых значений.

Возможно, что атомы могут исчезнуть, хотя это редкое явление. Два основных случая, когда это может произойти, это когда метаданные запоминают результат из предыдущего запуска, но соответствующий атом был вытеснен из временного пространства, и когда атом был сохранен на другом worker-е, который больше не отвечает на запросы. Реже возникает ошибка контрольной суммы, которая показывает, что сохраненные данные больше не являются действительными и должны быть отброшены.

Когда атом исчезает, прерывается thunk, который его запросил, и входит в режим восстановления:

1. Система проверяет наличие (но не контрольные суммы) всех других атомов, на которые ссылается вход thunk-а. Это происходит потому, что атомы, скорее всего, генерируются одновременно и на том же worker-е, и исчезновение атома коррелирует с исчезновением других атомов из примерно того же времени и места.
2. Система просматривает метаданные для ссылок на любые атомы, обнаруженные отсутствующими во время предыдущего шага. Это приведет к тому, что некоторые thunk-и вернутся из состояния “выполнено” в состояние “еще не выполнено”, потому что их результат был отброшен. Затем ядро обнаружит это и снова запланирует их выполнение.

Повторно выполненные thunk-и затем снова произведут атом, и выполнение может продолжиться.

Массивы атомов

Особенностью слоя атомов является способ, которым выполняются перестановки - первый слой из $M$ thunk-ов каждый производит несколько миллионов строк данных, а затем второй слой из $N$ thunk-ов каждый читает вывод предыдущего слоя для выполнения другой операции (обычно, некоторая форма сокращения), но каждая строка из первого слоя читается только одним thunk-ом из второго слоя.

Было бы очень расточительно, если бы каждый thunk во втором слое читал все данные из первого слоя (каждая строка была бы прочитана $N$ раз, из которых $N-1$ были бы ненужными), но именно это произошло бы, если бы каждый thunk из первого слоя производил ровно один атом.

С другой стороны, если каждый thunk в первом слое производит один атом для каждого thunk во втором слое, операция перемешивания будет включать $M\cdot N$ атомов в общей сложности - миллион атомов для $M = N = 1000$. Хотя накладные расходы на атомы не являются чрезмерными, добавление идентификатора атома, идентификатора арендатора, типа данных атома, размера и небольшого учета может достигать нескольких сотен байтов на атом. Хотя 100 МБ может показаться небольшой ценой, чтобы перемешать около 4 ГБ фактических данных, эти фактические данные находятся в слое атомов (который предназначен для больших данных), в то время как 100 МБ представляют собой значительную часть общего бюджета метаданных в размере 1,5 ГБ.

Чтобы обойти это, Envision поддерживает массивы атомов:

• Все атомы в массиве атомов записываются одновременно и хранятся вместе как в памяти, так и на диске.
• Зная идентификатор массива атомов, легко вывести идентификатор i-го атома в массиве.

Благодаря этому массив атомов имеет те же накладные расходы, что и один атом. В перемешивании thunk’и первого слоя будут производить $M$ массивов из $N$ атомов каждый. Thunk’и второго слоя будут запрашивать $M$ атомов, по одному из каждого массива, на позиции, соответствующей рангу этого thunk’а в перемешивании.

В заключение, несколько статистических данных о производстве! За час типичный работник выполнит 150 000 thunk’ов и запишет 200 000 атомов (массивы атомов учитываются только один раз), что составляет 750 ГиБ промежуточных данных.

В следующей и последней статье этой серии мы обсудим слои, которые позволяют распределенное выполнение происходить.

Бесстыдная реклама: мы нанимаем инженеров по программному обеспечению. Возможна удаленная работа.