Concurrencia y paralelismo en Elixir

Esta es la segunda parte de la serie de Percy Grunwald sobre Elixir. No te pierdas el primer artículo sobre coincidencia de Unicode en Elixir.

Los ejercicios en Exercism son pequeños, sintéticos y, a menudo, aparentemente triviales. Es fácil imaginar que quienes ya tienen experiencia no tendrían nada que aprender de ellos. Sin embargo, resolver estos problemas sintéticos puede llevarte a aprender y aplicar partes de tu lenguaje que quizás nunca exploraste. Este nuevo aprendizaje puede llevarte a resolver problemas del mundo real de forma más eficiente o más expresiva.

Parallel Letter Frequency es un ejercicio de dificultad media en el track de Elixir de Exercism que despliega una cantidad sorprendente de lecciones interesantes. Para resolver este problema con éxito, tu solución debe ejecutarse en paralelo en varios procesos worker. Lograr un verdadero paralelismo en Elixir es sorprendentemente fácil en comparación con otros lenguajes, pero si nunca has escrito código concurrente en Elixir, puede parecer un poco abrumador. Una de las cosas que descubrirás al resolver este ejercicio es lo fácil que Elixir hace escribir código que puede ejecutarse de forma concurrente o en paralelo. Aplicar estas habilidades a tu código puede tener un impacto importante en el rendimiento de tus aplicaciones.

Este ejercicio requiere que implementes una función, Frequency.frequency/2, que determine la frecuencia de letras en una lista de strings. El cálculo debe realizarse en varios procesos worker, determinados por el argumento workers:

iex> Frequency.frequency(["Freude", "schöner", "Götterfunken"], workers)
%{
  "c" => 1,
  "d" => 1,
  "e" => 5,
  ...
  "ö" => 2
}

En este artículo vamos a explorar la concurrencia en Elixir tomando una solución secuencial funcional de este ejercicio y haciéndola concurrente. Sin embargo, antes de meternos en el código, dediquemos un momento a examinar qué significan en realidad «concurrencia» y «paralelismo», y cómo lograr ambos en Elixir en comparación con otros lenguajes.

Concurrencia y paralelismo

Concurrencia y paralelismo son términos relacionados, pero no significan exactamente lo mismo. Un programa concurrente es aquel en el que varias tareas pueden estar «en progreso», pero, en un momento dado, solo una tarea se está ejecutando en la CPU (por ejemplo, ejecutar una tarea mientras otra espera E/S, como leer o escribir en el disco o en una red). Por otro lado, un programa paralelo es capaz de ejecutar varias tareas al mismo tiempo en varios núcleos de CPU.

Tanto la ejecución concurrente como la paralela pueden ofrecer aumentos de velocidad significativos, pero cuánto se puede acelerar (si es que se puede) depende de muchos factores. Hay casos en los que la concurrencia o el paralelismo no son siquiera posibles; puede que la tarea que intentas completar no se preste a una ejecución concurrente ni paralela, o que el runtime no las soporte. Si tu caso sí permite una ejecución concurrente o paralela, cuánta aceleración es posible depende en gran medida de si la tarea está limitada por E/S o por la CPU, y de si hay más de 1 núcleo de CPU disponible.

A pesar de la cantidad de factores que influyen, hay algunas «reglas generales» para determinar si la concurrencia o el paralelismo son posibles y cuánta aceleración cabe esperar. En primer lugar, la ejecución concurrente es posible en un solo núcleo de CPU, pero la paralela no. En segundo lugar, el paralelismo y la concurrencia deberían dar una aceleración significativa a las tareas limitadas por E/S, y la aceleración debería ser esencialmente la misma para ambas. Por último, las tareas limitadas por la CPU deberían tener el mismo rendimiento (o peor) cuando se ejecutan de forma concurrente, y, en general, la aceleración solo es posible cuando se ejecutan en paralelo en varios núcleos de CPU.

El cálculo de la frecuencia de letras en este ejercicio es un ejemplo de una tarea limitada por la CPU, así que, según las reglas generales anteriores, solo debería ser posible acelerar el cálculo haciéndolo en paralelo en varios núcleos de CPU.

Concurrencia y paralelismo en Elixir frente a otros lenguajes

Muchos lenguajes populares te dan herramientas para escribir código concurrente, pero lograr paralelismo suele ser mucho más complejo y estar lleno de compensaciones.

Por ejemplo, la concurrencia es de primera clase en JavaScript cuando se ejecuta en el runtime de Node.js, y las versiones predeterminadas de las funciones relacionadas con E/S son casi siempre «asíncronas» (es decir, concurrentes). Por ejemplo, fs.ReadFile es una función concurrente y la forma estándar de leer el contenido de un archivo. Esto es genial, pero también tiene sus desventajas en forma de infierno de callbacks o la necesidad de Promises. Además, como Node no reparte el tiempo de CPU por igual entre las tareas, todavía es posible bloquear la ejecución con una sola tarea intensiva en CPU. Paralelizar la ejecución en Node es posible, pero ciertamente no es fácil. Dado que Node es de un solo hilo, la única forma de lograr paralelismo es crear manualmente procesos worker con el módulo cluster o ejecutar varias instancias de tu programa e implementar manualmente la comunicación entre ellas.

Python, a diferencia de Node, no es concurrente por defecto, pero sí te ofrece varias herramientas para escribir código concurrente y paralelo. Sin embargo, cada opción tiene sus compensaciones y elegir una no es necesariamente sencillo. Podrías usar el módulo threading y lidiar con el hecho de que el global interpreter lock (GIL) limita la ejecución a un solo hilo a la vez, lo que hace imposible el paralelismo. La otra opción es el módulo multiprocessing de Python, que evita la limitación del GIL creando procesos del sistema operativo (en lugar de hilos). Usar multiprocessing hace posible el paralelismo, pero tiene la desventaja de que los procesos del sistema operativo tardan más en crearse y consumen más memoria que los hilos.

Escribir código concurrente y paralelo en Elixir es mucho más sencillo, ya que Elixir es concurrente a nivel de runtime. La reputación de Elixir por su enorme escalabilidad viene de que se ejecuta en la máquina virtual BEAM, que ejecuta todo el código dentro de «procesos» extremadamente ligeros que se ejecutan de forma concurrente dentro de la VM. Los procesos de BEAM tienen un costo de creación insignificante y usan cantidades mínimas de memoria en comparación con los hilos y procesos del sistema operativo que crean los módulos de concurrencia de Python. Además, en contraste con el bucle de eventos de Node, la VM de BEAM tiene planificadores que asignan el tiempo de CPU disponible a todos los procesos, lo que asegura que una sola tarea intensiva en CPU no pueda bloquear la ejecución de otros procesos.

Debido a esta arquitectura, pasar de ejecutar procesos de forma concurrente a ejecutarlos en paralelo es solo cuestión de añadir más núcleos de CPU. De hecho, desde hace muchos años BEAM activa automáticamente las capacidades de multiprocesamiento simétrico (SMP) en sistemas con varios núcleos, lo que permite que los planificadores de la VM asignen tiempo de CPU de todos los núcleos a los procesos en ejecución. En Elixir, no solo la concurrencia es de primera clase, sino que no hay distinción entre código concurrente y paralelo. Todo lo que necesitas hacer es escribir código concurrente y la VM lo paralelizará automáticamente y por defecto si hay más de 1 núcleo de CPU disponible.

Escribir código concurrente en Elixir con el módulo Task

Como se mencionó antes, en Elixir, la concurrencia se logra distribuyendo operaciones entre varios procesos de BEAM. Puedes crear procesos con mucha facilidad usando funciones como Kernel.spawn_link/1, pero es mucho mejor que uses las increíblemente potentes abstracciones que ofrece el módulo Task:

El caso de uso más común de [Task] es convertir código secuencial en código concurrente calculando un valor de forma asíncrona.

El módulo Task te permite escribir código concurrente increíblemente limpio en Elixir: sin infierno de callbacks y sin necesidad de Promises.

Para este ejercicio, Task.async_stream/3 es una gran opción:

async_stream(enumerable, function, options \\ [])

Task.async_stream/3 devuelve un stream que ejecuta la function dada de forma concurrente en cada elemento de enumerable. Por defecto, la cantidad de procesos creados (workers) es igual a la cantidad de elementos en enumerable. Esto nos da una forma directa de controlar el nivel de paralelismo con el argumento workers (suponiendo que tengamos suficientes núcleos de CPU). Todo lo que necesitamos hacer es dividir la lista de letras en la cantidad correcta de fragmentos y usar Task.async_stream/3 para procesar cada fragmento en un worker independiente.

Hacer concurrente la función secuencial de frecuencia de letras

Empecemos con una implementación secuencial funcional de mi solución:

def frequency(texts, _workers) do
  texts
  |> get_all_graphemes()
  |> count_letters()
end

defp get_all_graphemes(texts) do
  texts
  |> Enum.join()
  |> String.graphemes()
end

defp count_letters(graphemes) do
  Enum.reduce(graphemes, %{}, fn grapheme, acc ->
    if String.match?(grapheme, ~r/^\p{L}$/u) do
      downcased_letter = String.downcase(grapheme)
      Map.update(acc, downcased_letter, 1, fn count -> count + 1 end)
    else
      acc
    end
  end)
end

Podemos hacer concurrente la implementación anterior así:

1. Dividir la lista de grafemas que devuelve get_all_graphemes/1 en una cantidad de fragmentos igual a la cantidad de workers
2. Procesar cada fragmento con count_letters/1 en un worker usando Task.async_stream/3
3. Combinar los resultados de cada worker en un solo resultado

Aquí tienes un diagrama de los pasos anteriores:

Solo necesitamos implementar 2 nuevas funciones auxiliares para habilitar la lógica concurrente: una para dividir los grafemas en fragmentos (split_into_chunks/2) y otra para combinar el stream de resultados de los workers (merge_results/1). Aquí tienes una forma de implementar esas funciones auxiliares:

defp split_into_chunks(all_graphemes, num_chunks) do
  all_graphemes_count = Enum.count(all_graphemes)
  graphemes_per_chunk = :erlang.ceil(all_graphemes_count / num_chunks)

  Enum.chunk_every(all_graphemes, graphemes_per_chunk)
end

defp merge_results_stream(results_stream) do
  Enum.reduce(results_stream, %{}, fn {:ok, worker_result}, acc ->
    Map.merge(acc, worker_result, fn _key, acc_val, worker_val ->
      acc_val + worker_val
    end)
  end)
end

Con esas 2 funciones implementadas, lo único que queda por hacer para que la función frequency/2 se ejecute de forma concurrente es agregar llamadas a las nuevas funciones auxiliares y reemplazar la llamada directa a count_letters/1 por Task.async_stream/3:

def frequency(texts, workers) do
  texts
  |> get_all_graphemes()
  |> split_into_chunks(workers)
  |> Task.async_stream(&count_letters/1)
  |> merge_results_stream()
end

La función anterior es una implementación concurrente totalmente funcional y pasa todas las pruebas. A pesar de ser concurrente, el código se lee exactamente como código secuencial normal, lo que demuestra el poder de las abstracciones que ofrece Task.

¿La versión concurrente es realmente paralela?

Como mencioné antes en el artículo, no hay distinción entre código concurrente y paralelo en Elixir. Si establecemos workers en un número mayor que 1, y tenemos más de 1 núcleo de CPU disponible, la VM de BEAM paraleliza automáticamente la ejecución de los procesos creados.

Por defecto, BEAM inicia un planificador por cada núcleo de CPU (lógico) disponible. Puedes consultar la cantidad de planificadores que BEAM ha iniciado con :erlang.system_info/1:

iex> :erlang.system_info(:schedulers_online)
8

Esto representa la cantidad máxima de procesos de la VM que pueden estar ejecutándose al mismo tiempo. Establecer workers en un número mayor que la cantidad de planificadores no aumenta el paralelismo y puede perjudicar el rendimiento.

Conclusión

Convertir el código de secuencial a concurrente resulta ser mucho más fácil de lo esperado usando el módulo Task de Elixir. El código concurrente añade algo de complejidad extra al dividir la lista de grafemas y combinar los resultados de los workers, pero el resultado final es sorprendentemente limpio.

Antes de resolver este problema de Exercism había oído hablar de Task, pero nunca lo había usado. Después de aplicarlo en mi solución, ahora lo consideraría una parte indispensable de mi caja de herramientas de Elixir.

Podrías usar esta nueva herramienta de muchas formas para intentar optimizar el rendimiento de tus aplicaciones. La mayoría de las aplicaciones web están limitadas por E/S y, por lo tanto, se beneficiarían de la concurrencia incluso si solo hay un único núcleo de CPU disponible. Una forma bastante confiable de acelerar una aplicación web es hacer solicitudes HTTP de forma concurrente:

def call_apis_async() do
  ["https://api.example.com/users/123", ...]
  |> Task.async_stream(&HTTPoison.get/1)
  |> Enum.into([], fn {:ok, res} -> res end)
end

El código anterior aplica Task.async_stream/3 para llamar a todas las URL de la lista de forma concurrente, en lugar de esperar a que cada solicitud se complete antes de iniciar la siguiente, lo que debería dar una aceleración significativa según la duración de cada solicitud.