Envision VM (parte 3), Átomos y Almacenamiento de Datos

Victor Nicollet

Este artículo es el tercero de una serie de cuatro partes sobre el funcionamiento interno de la máquina virtual Envision: el software que ejecuta los scripts de Envision. Ver parte 1, parte 2 y parte 4. Esta serie no cubre el compilador de Envision (quizás en otra ocasión), así que supongamos que el script ha sido de alguna manera convertido al bytecode que la máquina virtual Envision utiliza como entrada.

Durante la ejecución, los thunks leen datos de entrada y escriben datos de salida, a menudo en grandes cantidades.

• Mil millones de booleanos (un bit por valor) ocupan 125MB.
• Mil millones de números de punto flotante (precisión de 32 bits) ocupan 4GB.
• Mil millones de líneas mínimas de ventas (fecha, ubicación, EAN-13, cantidad) ocupan entre 14GB y 33GB (¡o más!) dependiendo de cómo se codifiquen los valores.

Esto crea dos desafíos: cómo preservar estos datos desde el momento en que se crean y hasta que se usan (parte de la respuesta es: en unidades NVMe distribuidas en varias máquinas), y cómo minimizar la cantidad de datos que pasan por canales más lentos que la RAM (red y almacenamiento persistente).

Capa de Metadatos

Una parte de la solución consiste en tener dos capas de datos separadas, con los datos siendo enviados a una de las dos capas según su naturaleza. La capa de metadatos contiene información sobre los datos reales, y sobre los scripts que se están ejecutando:

• Cuando un thunk devuelve datos con éxito, el identificador único de esos datos se guarda en esta capa.
• Cuando un thunk falla, los mensajes de error producidos por el thunk se guardan en esta capa.
• Cuando un thunk ha devuelto un nuevo thunk (y el DAG de sus padres), el DAG serializado se guarda en esta capa.
• Un thunk puede guardar puntos de control en la capa de metadatos (usualmente consistentes en el identificador de un bloque de datos); si un thunk se interrumpe antes de completarse, puede cargar su punto de control desde la capa de metadatos y reanudar su trabajo desde esa posición.

En otras palabras, la capa de metadatos puede verse como un diccionario que asigna thunks a resultados, donde la naturaleza exacta del resultado depende de lo que el thunk realmente devolvió.

La capa de metadatos también puede contener información adicional sobre la estructura de los datos referenciados. Por ejemplo, si un thunk ha devuelto un par de vectores, entonces la capa de metadatos contendrá el identificador único de cada vector. Esto permite a los consumidores acceder a un vector sin tener que cargar ambos.

Hay dos límites en los valores almacenados en la capa de metadatos: una entrada no puede exceder los 10MB (¡por lo que un DAG serializado tampoco puede exceder esta cantidad!), y el espacio total de almacenamiento para la capa de metadatos es de 1.5GB. Usualmente, hay alrededor de un millón de valores en esta capa, para un tamaño promedio de entrada de 1.5KB.

La capa de metadatos siempre reside en la RAM para garantizar un acceso rápido. Actúa como la fuente de verdad para la ejecución de thunks: un thunk se ha ejecutado si, y solo si, hay un resultado asociado a ese thunk en la capa de metadatos—aunque esto no garantiza que los datos referenciados por ese resultado estén disponibles.

Cada trabajador en el clúster mantiene su propia copia de la capa de metadatos. El trabajador transmite cada cambio a esta capa (provocado por la ejecución de thunks locales) a todos los demás trabajadores en el clúster, y también al planificador. Esto se hace sobre una base de «mejor esfuerzo»: si un mensaje de transmisión no llega a su destino, se descarta¹ sin un reintento.

Cada segundo, la capa de metadatos se persiste en disco, de manera incremental. En caso de falla o reinicio, el trabajador tardará uno o dos segundos en recargar toda la capa desde el disco para recordar lo que estaba haciendo.

Mantener grandes bases de datos en memoria

Como se mencionó anteriormente, la capa de metadatos puede contener un millón de entradas. Cada DAG individual puede contener cientos de miles de nodos. Todos ellos tienen vidas largas—desde minutos hasta horas. Mantener millones de objetos de larga duración en la memoria es bastante difícil para el recolector de basura de .NET.

La recolección de basura en .NET es un tema complejo (aunque existe una excelente serie de Konrad Kokosa para profundizar en los detalles de bajo nivel), pero el problema general es una combinación de tres hechos:

• El costo en rendimiento de una pasada de recolección de basura es proporcional al número de objetos vivos en la zona de memoria que se está procesando. Procesar millones de objetos, a menudo con miles de millones de referencias entre ellos, llevará al recolector de basura varios segundos en procesarlos.
• Para evitar pagar este costo, el recolector de basura de .NET trabaja con áreas separadas de memoria, llamadas generaciones, según la antigüedad de los objetos en ellas. La generación más joven, Gen0, se somete a recolección de basura con frecuencia pero solo contiene objetos asignados desde la última pasada (por lo que, solo unos pocos). La generación más vieja, Gen2, solo se recolecta si tanto Gen1 como Gen0 fueron recolectadas pero no se obtuvo memoria libre suficiente. Esto será bastante raro mientras la mayoría de las asignaciones de objetos sean pequeñas y de corta duración.
• Sin embargo, una operación normal de un thunk implica grandes arreglos de valores, que se asignan en el Large Object Heap, un área separada de Gen0, Gen1 y Gen2. Cuando el Large Object Heap se queda sin espacio, se realiza una recolección de basura completa, que también recoge Gen2.

Y Gen2 es donde se sitúan los millones de objetos de los DAGs y de la capa de metadatos.

Para evitar esto, hemos construido tanto los DAGs como la capa de metadatos para utilizar solo muy pocos objetos.

Cada DAG consiste en solo dos asignaciones—un arreglo de nodos y un arreglo de aristas, ambos de los cuales son tipos de valor no administrados, de modo que el GC ni siquiera necesita recorrer su contenido para seguir las referencias que puedan contener. Cuando se necesita un thunk para ser ejecutado, se deserializa de la representación binaria del DAG², que está presente en la capa de metadatos.

La capa de metadatos tiene contenidos de longitud variable, por lo que se construye extrayendo trozos de un gran byte[], utilizando ref struct y MemoryMarshal.Cast para manipular los datos sin copiarlos.

Espacio de Trabajo

Un clúster tiene entre 512GiB y 1.5TiB de RAM, y entre 15.36TB y 46.08TB de almacenamiento NVMe. La mayor parte de este espacio se dedica a almacenar los resultados intermedios de la evaluación de thunks.

La RAM es un recurso valioso: representa solo el 3% del espacio de almacenamiento disponible, pero es entre 100 y 1000 veces más rápida para leer y escribir. Hay un beneficio significativo en asegurar que los datos que están a punto de ser leídos por un thunk ya estén presentes en la memoria (o que nunca hayan salido de la memoria en primer lugar).

Además, es casi imposible utilizar el 100% de la RAM disponible en .NET— el sistema operativo tiene necesidades de memoria variables, y no dispone de una forma confiable de comunicar al proceso de .NET que debe ceder algo de memoria, lo que resulta en que el proceso sea terminado por oom (fuera de memoria).

Envision resuelve este problema delegando la gestión de las transferencias de RAM a NVMe al sistema operativo. Hemos liberado este código como Lokad.ScratchSpace. Esta biblioteca mapea en memoria todo el espacio de almacenamiento disponible en las unidades NVMe, y lo expone como un almacén de blobs que la aplicación puede utilizar para:

1. escribir bloques de datos (hasta 2GB cada uno) en el espacio de trabajo, ya sea directamente o serializando desde un objeto administrado. Esta operación devuelve un identificador de bloque.
2. leer bloques de datos utilizando sus identificadores. Esta operación fija el bloque y lo expone a la aplicación como un ReadOnlySpan<byte>, que la aplicación debería luego copiar (o deserializar) a la memoria administrada.

Una vez que el espacio de trabajo se llena, se descartan los bloques más antiguos para hacer espacio para nuevos datos. Esto significa que es posible que una operación de lectura falle, si el identificador apunta a un bloque que ha sido descartado, pero esto es algo raro durante la ejecución de un script de Envision—raramente una sola ejecución produce decenas de terabytes. Por otro lado, esto puede impedir que una nueva ejecución reutilice los resultados de una ejecución anterior.

La clave para utilizar un espacio de trabajo mapeado en memoria es que la RAM disponible se distribuye entre tres tipos de páginas³: memoria que pertenece a procesos (como el proceso .NET de Envision), memoria que es una copia exacta, byte por byte, de una porción de un archivo en disco, y memoria que está destinada a ser escrita en un archivo en disco.

Cuando la memoria es una copia de un archivo en disco, en cualquier momento puede ser liberada por el sistema operativo y utilizada para otro propósito—para ser asignada a un proceso para su propio uso, o para convertirse en una copia de otra porción de un archivo en disco. Aunque no es instantáneo, estas páginas actúan como un búfer de memoria que puede reasignarse rápidamente para otro uso. Y hasta que sean reasignadas, el sistema operativo sabe que contienen una copia de una región específica de memoria persistente, por lo que cualquier solicitud de lectura para esa región se redirigirá a la página existente, evitando así cargar desde el disco.

La memoria que está destinada a ser escrita en disco, eventualmente será escrita y se convertirá en una copia de la región donde fue escrita. Esta conversión está limitada por la velocidad de escritura de las unidades NVMe (en el orden de 1GB/s).

La memoria asignada al proceso no puede convertirse de nuevo a los otros dos tipos sin ser liberada por el proceso (lo cual el GC de .NET a veces hará, después de que una colección haya liberado una gran cantidad de memoria). Toda la memoria asignada a través de .NET, incluidos todos los objetos administrados y todo lo que supervisa el GC, debe pertenecer a este tipo de memoria.

En un trabajador típico, el 25% de la memoria se asigna directamente al proceso .NET, el 70% es una copia de solo lectura de regiones de archivos, y el 5% está en proceso de ser escrito.

Capa de Átomos

El principio general es que cada thunk escribe su salida en el espacio de trabajo como uno o más átomos, y luego almacena los identificadores de esos átomos en la capa de metadatos. Los thunks subsiguientes luego cargan estos identificadores desde la capa de metadatos, y los utilizan para consultar el espacio de trabajo en busca de los átomos que necesitan.

El nombre «Átomo» se eligió porque no es posible leer solo una parte de un átomo: solo se pueden recuperar en su totalidad. Si una estructura de datos necesita permitir solicitar solo una parte de su contenido, en su lugar lo guardamos como múltiples átomos, que luego pueden recuperarse de manera independiente.

Algunos átomos están comprimidos; por ejemplo, la mayoría de los vectores booleanos no se representan como bool[], que consume un byte por elemento, sino que se compactan a 1 bit por valor, y luego se comprimen para eliminar largas secuencias de valores idénticos.

Es posible que los átomos desaparezcan, aunque esto es una ocurrencia rara. Las dos situaciones principales en las que esto puede suceder son cuando la capa de metadatos recuerda un resultado de una ejecución anterior, pero el átomo correspondiente fue expulsado del espacio de trabajo mientras tanto, y cuando el átomo se almacenó en un trabajador diferente que ya no responde a las solicitudes. Con menos frecuencia, un error de suma de verificación revela que los datos almacenados ya no son válidos y deben descartarse.

Cuando un átomo desaparece, el thunk que lo solicitó se interrumpe y entra en modo de recuperación:

1. El sistema verifica la presencia (pero no las sumas de verificación) de todos los demás átomos referenciados por las entradas del thunk. Esto se debe a que es probable que los átomos se generen al mismo tiempo y en el mismo trabajador, y la desaparición de un átomo está correlacionada con la desaparición de otros átomos de alrededor del mismo momento y lugar.
2. El sistema examina la capa de metadatos en busca de referencias a cualquiera de los átomos descubiertos como faltantes durante el paso anterior. Esto hará que algunos thunks pasen de estar “ejecutados” a “aún no ejecutados” porque su resultado fue descartado. Luego, el kernel detectará esto y los reprogramará.

Los thunks re-ejecutados producirán el átomo nuevamente, y la ejecución puede reanudarse.

Arreglos de Átomos

Un aspecto particular de la capa de átomos es la forma en la que se realizan los shuffles—una primera capa de $M$ thunks cada uno produce varios millones de líneas de datos, y luego una segunda capa de $N$ thunks lee la salida de la capa anterior para realizar otra operación (usualmente, alguna forma de reduce), pero cada línea de la primera capa es leída únicamente por un thunk de la segunda capa.

Sería muy derrochador que cada thunk de la segunda capa leyera todos los datos de la primera capa (cada línea se leería $N$ veces, de las cuales $N-1$ serían innecesarias), pero esto es exactamente lo que sucedería si cada thunk de la primera capa produjera exactamente un átomo.

Por otro lado, si cada thunk en la primera capa produce un átomo por cada thunk en la segunda capa, la operación de shuffle involucrará $M\cdot N$ átomos en total—un millón de átomos para $M = N = 1000$. Aunque el sobrecoste en átomos no es excesivo, al sumar un identificador de átomo, identificador de tenant, tipo de dato del átomo, tamaño y un poco de contabilidad, aún puede llegar a alcanzar unos pocos cientos de bytes por átomo. Aunque 100MB puede parecer un precio pequeño a pagar para mover 4GB de datos reales, esos datos reales residen en la capa de átomos (que está diseñada para datos grandes), mientras que 100MB representa una parte considerable del presupuesto total de 1,5GB de la capa de metadatos.

Para solucionar esto, Envision soporta arrays de átomos:

• Todos los átomos en un array de átomos se escriben al mismo tiempo, y se mantienen juntos tanto en la memoria como en el disco.
• Dado el identificador del array de átomos, es fácil derivar el identificador del i-ésimo átomo en el array.

Gracias a esto, un array de átomos tiene el mismo sobrecoste que un solo átomo. En un shuffle, los thunks de la primera capa producirían $M$ arrays de $N$ átomos cada uno. Los thunks de la segunda capa solicitarían cada uno $M$ átomos, uno de cada array, en la posición correspondiente al rango de ese thunk en el shuffle.

Para concluir, ¡algunas estadísticas de producción! En una hora, un trabajador típico ejecutará 150 000 thunks y escribirá 200 000 átomos (los arrays de átomos se cuentan solo una vez) que representan 750GiB de datos intermedios.

En el siguiente y último artículo de esta serie, discutiremos las capas que permiten que se lleve a cabo la ejecución distribuida.

Anuncio sin complejos: estamos contratando ingenieros de software. El trabajo remoto es posible.