Pacemaking

El problema

Imaginá que hay un número $N$ de servidores en una red P2P completamente conectada. Estos servidores pueden copiarse archivos entre sí. En este sistema imaginario, hay una única máquina que cada $δ$ minutos produce un archivo que todas las demás máquinas necesitan obtener (por ejemplo, un checkpoint de base de datos), y ese archivo resulta ser relativamente grande —digamos > 200 GB, para dar una idea.

Los consumidores del archivo necesitarían enterarse de su existencia y luego descargarlo. Podríamos imaginar un sistema de notificaciones push, o algún tipo de sistema de polling, pero para simplificar supongamos que el productor almacena un registro de la existencia de ese archivo en un lugar compartido. Entonces, digamos que cada $1 m$ , cada servidor verificará ese lugar compartido para ver si hay un archivo nuevo para descargar. Si lo hay, lo descargará desde cualquiera de las máquinas que ya lo tengan. Si no lo hay, no hará nada.

El desafío

Este tipo de patrón de consumo es propenso a ataques de denegación de servicio contra el productor del archivo: si el productor es la única máquina que tiene el archivo, es probable que muchas máquinas intenten descargarlo al mismo tiempo, lo que podría fácilmente saturar al productor. Este es un ejemplo clásico de un «punto caliente» en un sistema distribuido.

Algo que vale la pena notar sobre este sistema es que casi siempre terminará estresando al productor:

Las máquinas suelen iniciarse aproximadamente al mismo tiempo (debido a los ciclos de despliegue), lo que significa que tenderán a acumularse contra el productor en grupos.
Un subconjunto de máquinas puede ser suficiente para saturar al productor (y en la práctica a menudo lo es). Una vez que el productor está saturado y lento, más máquinas se suman, agravando el problema. Esto es un bucle de retroalimentación positiva que puede fácilmente derivar en un ataque de denegación de servicio.
Una alternativa aquí es algo en la forma de una «tormenta de reintentos»: el productor está lento, provoca timeouts (o directamente fallos), y las máquinas reintentan, empeorando el problema.

Esta no es una situación agradable si querés dormir tranquilo por las noches (¡preguntame cómo lo sé!). Entonces, ¿qué podemos hacer al respecto?

Lo que realmente tenemos entre manos es un thundering herd: un único productor a punto de ser arrasado por todos los consumidores a la vez. El balanceo de carga clásico no aplica, ya que al inicio solo hay un servidor con el archivo. Las defensas reactivas como límites de tasa, colas o backoff solo entran en juego una vez que la estampida ya comenzó, momento en el cual el productor ya está saturado.

Lo que hay que notar es que los consumidores también son fuentes futuras: una vez que una máquina tiene el archivo, puede servirlo a sus pares a través del mismo tejido P2P. Entonces, si podemos escalonar la manada en oleadas, el conjunto de réplicas crece a medida que cada oleada se completa, y las oleadas posteriores pueden obtener el archivo de sus pares en lugar del productor.

Hay muchas formas de hacer esto; por ejemplo, el productor podría enviar proactivamente el archivo a futuros consumidores antes de anunciarlo (garantizando que exista un pool de máquinas en el momento en que se anuncia), o podría anunciar el archivo solo a un subconjunto de máquinas, y luego hacer que esas máquinas lo anuncien a otras, etc.

El problema con la mayoría de estas ideas es que son relativamente costosas en términos de implementación (por ejemplo, ahora necesitamos implementar ese mecanismo que nos permite enviar archivos proactivamente), y todos sabemos que a nadie le gustan los PRs grandes con alto costo de mantenimiento. Ya tenemos una red de distribución de contenido en funcionamiento y un lugar para anunciar la existencia del archivo, así que ¿podemos arreglarnos con lo que hay?

La idea clave detrás del pacemaking es esta: podemos manipular la probabilidad de que cualquier máquina dada descargue un archivo en cualquier período de tiempo determinado, con el fin de garantizar que la carga sobre el productor se disipe con el tiempo al agregar más réplicas de forma controlada. Vamos a profundizar en esto.

Distribuyendo las descargas en el tiempo

Tenemos $N$ servidores a los que queremos distribuir el archivo con el paso del tiempo. Una forma de distribuir la carga en el tiempo es organizar las descargas de modo que las máquinas que descarguen el archivo en un momento dado sean aproximadamente el mismo número que las máquinas que ya lo tienen. Para hacer eso, imaginemos que dividimos el tiempo en buckets:

En $T = 0$ , tenemos exactamente 1 máquina (el productor) que tiene el archivo, así que queremos que como máximo 1 máquina descargue el archivo.
En $T = 1$ (asumiendo que todo salió bien) tenemos 2 máquinas que tienen el archivo, así que queremos que como máximo 2 máquinas descarguen el archivo.
En $T = 2$ (asumiendo que todo salió bien), tenemos 4 máquinas que tienen el archivo, así que queremos que como máximo 4 máquinas descarguen el archivo.

Esencialmente, al comienzo del $T$ -ésimo bucket de tiempo, queremos que como máximo $2^{T}$ máquinas descarguen el archivo, ya que $2^{max (T - 1, 1)}$ máquinas ya lo habrán descargado en ese momento. Al final del $T$ -ésimo bucket de tiempo, $2^{T + 1}$ máquinas habrán descargado el archivo. Como tenemos $N$ máquinas que necesitan descargar el archivo, necesitamos $⌈ lo g_{2} (N)⌉$ buckets en el tiempo para asegurarnos de que todas las máquinas lo hayan descargado; y el último bucket tendrá $N - 2^{⌊ l o g_{2} (N)⌋}$ máquinas descargando el archivo.

Un gráfico de eso luce más o menos así:

Hay algunas cosas que vale la pena notar aquí:

En realidad no tenemos una forma de distribuir las descargas en el tiempo de modo que se ajusten a esta distribución.
Estamos asumiendo que todas las descargas de archivos son exitosas.
Hemos asumido convenientemente que la duración de la descarga del archivo cabe dentro del bucket, de modo que al final del bucket las máquinas habrán descargado el archivo.
El último bucket está destinado a tener la mayor cantidad de descargas concurrentes, por lo que la mayoría de las máquinas tendrán que esperar esencialmente $(B - 1) S$ hasta que puedan descargar el archivo, donde $B$ es el número de buckets de tiempo y $S$ es su tamaño.

Trabajaremos estos detalles en las próximas secciones.

Asignando máquinas a buckets de tiempo

Para lograr esta distribución de descargas, nos gustaría asignar máquinas a buckets de tiempo de tal manera que el número de máquinas asignadas a cada bucket sea igual al número de máquinas que han descargado el archivo en ese momento. Una forma de hacerlo es muestrear desde una distribución que tenga esta propiedad.

Entonces, imaginemos que $X$ es una variable aleatoria discreta que representa un número de bucket, con soporte sobre ${0, 1, \dots, B - 1}$ , donde $B$ es el número de bucket. Nos gustaría realmente asegurar algo similar a $P (X = i + 1) = 2 P (X = i)$ .

Una forma de hacerlo es pedir que $P (X = i) = 2^{i} q$ (lo cual satisfaría trivialmente la propiedad que realmente nos gustaría tener), y luego podemos resolver para $q$ , ya que debe ser una distribución de probabilidad:

i = 0 \sum B - 1 P (X = i) i = 0 \sum B - 1 2^{i} q q i = 0 \sum B - 1 2^{i} q (2^{B} - 1) q = 1 = 1 = 1 = 1 = \frac{1}{2 ^{B} - 1}

Lo que nos deja con $P (X = i) = \frac{2 ^{i}}{2 ^{B} - 1}$ . Ahora tenemos algo de matemática sofisticada que nos dice cómo distribuir las máquinas entre buckets; solo que ahora necesitamos muestrear desde esta distribución de alguna manera.

El muestreo por transformada inversa es un método para muestrear desde una distribución dada su función de distribución acumulada (CDF). La CDF de esta distribución está dada por:

F (x) = P (X \leq x) = i = 0 \sum x P (X = i) = i = 0 \sum x \frac{2 ^{i}}{2 ^{B} - 1} = \frac{2 ^{x + 1} - 1}{2 ^{B} - 1}

y la inversa de esa función es:

F^{- 1} (y) = argmin_{x} (\frac{2 ^{x + 1} - 1}{2 ^{B} - 1} \geq y) = argmin_{x} (2^{x + 1} - 1 \geq y (2^{B} - 1)) = argmin_{x} (2^{x + 1} \geq y (2^{B} - 1) + 1) = argmin_{x} (x + 1 \geq lo g_{2} (y (2^{B} - 1) + 1)) = ⌊ lo g_{2} (y (2^{B} - 1) + 1)⌋

Entonces, el muestreo por transformada inversa nos indica:

Muestrear $y$ desde una distribución uniforme entre 0 y 1.
Calcular $x = ⌊ lo g_{2} (y (2^{B} - 1) + 1)⌋$ .

Esto nos dará un bucket de muestra $x$ desde la distribución que queremos, que luce así:

Por fin, el algoritmo

Este algoritmo tiene dos lados: el productor del archivo y el consumidor del archivo. Del lado del productor, mantendremos todo como está, pero agregaremos algunos metadatos al archivo:

La hora de creación del archivo $T_{current}$ .
La hora de creación esperada para el próximo archivo $T_{next}$ , si se conoce.

Del lado del consumidor, lo fundamental es esto: es posible que hagamos polling del archivo varias veces antes de que se cree uno nuevo, así que necesitamos asegurarnos de que cualquiera sea el bucket en el que estamos, sea consistente a lo largo del tiempo.

Entonces, sabemos que:

Tenemos $N$ máquinas que necesitan descargar los datos.
Deben distribuirse en $B = ⌈ lo g_{2} (N)⌉$ buckets.
Necesitamos muestrear desde la distribución que derivamos arriba para determinar en qué bucket estamos.
Necesitamos asegurarnos de que el bucket en el que estamos sea consistente con el tiempo, cada vez que hagamos polling hasta que se cree el próximo archivo.
Tenemos $δ = T_{next} - T_{current}$ hasta que se cree el próximo archivo.

La pieza que falta aquí es determinar el tamaño $S$ de los buckets. Esto puede hacerse de cualquier número de maneras, como $S = \frac{\frac{1}{2} δ}{B}$ , o mediante algún tipo de estimación del tiempo de descarga. Lo importante es que $T_{current} + S B < T_{next}$ , de modo que todas las máquinas terminen de descargar antes de que el próximo archivo esté disponible.

Ahora que tenemos todo esto, necesitamos determinar en qué bucket estamos. El problema es que si seguimos el método de muestreo por transformada inversa, obtendremos un bucket diferente cada vez que hagamos polling de un nuevo archivo. Esto no es lo que queremos, ya que deseamos que a la máquina se le asigne el mismo bucket cada vez que verifique, al menos hasta que haya un nuevo archivo disponible para descargar.

Una forma sencilla de hacer esto es usar hashing. Podemos hashear la hora de creación del archivo y el nombre de la máquina para obtener un número que esté distribuido de forma aproximadamente uniforme (como en $r = hash (fileCreationTime, machineName)$ ), y luego usar ese número para determinar el bucket estableciendo $b = F^{- 1} (r)$ . De esta manera, a la máquina siempre se le asignará el mismo bucket hasta que se cree un nuevo archivo.

El resto del algoritmo es directo. Cada vez que verificamos si hay un archivo nuevo, hacemos lo siguiente:

Si no hay archivo nuevo, no hacemos nada.
De lo contrario, generamos $r$ y muestreamos el bucket actual $b = F^{- 1} (r)$ . Procedemos a descargar el archivo solo si $T_{next} \geq now \geq T_{current} + S b$

Lo que se dejó debajo de la alfombra

Como se mencionó antes, asumimos que todas las descargas de archivos son exitosas y que el tiempo de descarga cabe dentro del bucket. En mi escenario de producción particular, tenía $1 m$ buckets con $δ = 15 m$ , y las descargas ciertamente cabían dentro de ese bucket, así que no tuve que preocuparme por eso. Si tenés un escenario diferente, es posible que debas ajustar el tamaño del bucket en consecuencia.

El otro problema es que asumimos que todas las descargas de archivos son exitosas. Esto es algo problemático, ya que no tenemos forma de garantizarlo. Debido a detalles particulares del sistema de producción con el que estaba trabajando, esto en la práctica no importaba porque la probabilidad de que fallara una descarga era muy baja. Sin embargo, esto puede tenerse en cuenta en cierta medida modificando la distribución para considerar la probabilidad de que falle una descarga, y luego muestreando desde esa distribución actualizada.

Conclusión

Este algoritmo es una forma sencilla de garantizar que la carga sobre un productor de archivos se distribuya en el tiempo. Obviamente no es perfecto y tiene una serie de supuestos subyacentes, pero tiene la ventaja de que puede implementarse en unas 10 líneas de código y no requiere mucha infraestructura. Es un truco útil que conviene tener a mano si alguna vez te encontrás en una situación en la que necesitás distribuir carga en el tiempo, o más probablemente como inspiración si alguna vez te encontrás en una situación similar.

He ejecutado este algoritmo en producción y ha funcionado sin problemas durante lo que ahora son varios años (y de hecho, el suavizado que introdujo eliminó varias formas en que el sistema solía fallar).