AWS Lambda y Multithreads

Recientemente debí atender un caso muy particular en una carga de trabajo que corre en un entorno serverless en AWS.

En este escenario, se tenía un API Gateway que exponía -entre otros- un servicio para realizar la recepción de cierta información en formato JSON. Este servicio estaba desplegado en Lambda con un runtime de Java 11.

Este API por consideraciones muy particulares debía ser sincrónico. El lambda tomaba ese payload y realizaba una serie de validaciones por cada elemento de la colección que recibía y normalmente su tiempo de respuesta estaba en el rango de un par de segundos y en el margen del SLA que se solicitaba para el mismo.

El problema se daba cuando se enviaba un payload conformado por miles de líneas a validarse. Este procesamiento en el Lambda demoraba poco más de 35 segundos y en API Gateway, el máximo timeout permitido es de 29 segundos. Además, es un hard limit por lo cual no se puede exceder bajo ninguna consideración.

Existen diversas maneras de atender un caso como este:

• Las medidas que implican un rediseño para que el API sea asincrónico y que el emisor reciba un callback o bien consulte el estado del proceso posteriormente. Alternativa que por las consideraciones específicas no era viable.

• Otra posibilidad explorada es dar mayor capacidad de memoria al Lambda y por consiguiente mayor poder de cómputo para procesar la petición. Este escenario fue ejecutado y ciertamente se lograba dar respuesta en los límites de los 29 segundos.

Sin embargo, la solución definitiva fue recurrir a los elementos de Multithreading que nos brinda Java y por tanto paralelizar el procesamiento dentro del mismo Lambda.

¿Qué es un Multithreading?

Antes de ver como se emplean múltiples hilos o multithreading en Lambda, debemos recordar que en Java se refiere a la capacidad de un programa para realizar múltiples tareas simultáneamente dentro de un sólo proceso. Esto se logra mediante la creación de múltiples hilos de ejecución que pueden ejecutarse de manera concurrente dentro del mismo programa.

Un hilo es un proceso ligero que se ejecuta de forma independiente de otros hilos y tiene su propia pila de ejecución. Cada hilo puede realizar una tarea diferente, lo que permite que el programa realice múltiples operaciones al mismo tiempo.

Java proporciona soporte incorporado para esto a través de la clase java.lang.Thread y el paquete java.util.concurrent. La clase Thread proporciona métodos para crear, iniciar y controlar hilos, mientras que el paquete java.util.concurrent proporciona abstracciones de nivel superior para administrar operaciones concurrentes.

Esto se puede utilizar para mejorar el rendimiento y la capacidad de respuesta de las aplicaciones, especialmente en casos en los que hay operaciones de larga duración o tareas que se pueden realizar de manera concurrente.

Sin embargo, también puede introducir complejidad adicional y posibles problemas como condiciones de carrera, bloqueos y seguridad de hilos. Por lo tanto, debemos ser prudentes y muy cuidadosos en su uso, aplicando las mejores prácticas para la seguridad y sincronización de hilos.

Como podemos ir deduciendo, el uso de hilos tiene mucho sentido para el caso de uso que tenemos por delante.

Implementación

La Implementación del cambio implica crear una serie de Threads, cada uno a cargo de procesar un subset de los datos que se reciben. Simplifique el caso real en favor de facilitar su lectura.

  // supongamos que THREAD_NUMBER es el número de threads a usar

  for (int i = 0; i < THREAD_NUMBER; i++) {
       // subLista es una función que nos genera un subset del payload separado
       // para cada Thread. Esto para que cada uno trabaje con sus datos en
       // paralelo
       new Thread(new ValidatorThread(i, subLista(i, payload))).start();
   }

y la clase ValidatorThread sería esta:

private class ValidatorThread implements Runnable {

  private final int threadNumber;
  private final List<MyData> data;
  public ValidatorThread(int i, List<MyData> data) {
     this.threadNumber = i;
     this.data = data;
  }
  ...
  @Override
  public void run() {
    // ejecutamos nuestra lógica de negocio con base a
    // los datos que tenemos
  }

}

Si ejecutamos nuestro Lambda tal cual; veríamos un comportamiento no deseado. El lambda retorna justo después de que lanzamos los Threads, sin dar tiempo a que terminen de procesar.

Se preguntarán: ¿porqué?

La respuesta es sencilla, el flujo de ejecución es lineal: lanzamos los threads por medio del método start y nuestro código no espera a que concluyan para a su vez retornar con el proceso completado.

CountDownLatch

A nuestro rescate entra un CountDownLatch, el cual es un elemento de sincronización provisto en Java que permite que uno o más Threads esperen a que concluya un cierto número de operaciones antes de continuar. Podemos verlo como una barrera de sincronización.

Un CountDownLatch es inicializado con un valor, que especifica el número de veces que el método await() debe invocarse antes de que cualquier Thread en espera sea liberado. Cada vez que llamamos al método countDown() ese contador decrece. Cuando alcanza el valor de cero, todos los threads en espera son liberados y se puede proceder.

Un CountDownLatch es muy útil en Java cuando tenemos Threads que deben esperar a que otros Threads terminen antes de continuar. Lo debemos ver como un coordinador en la ejecución de un grupo de threads de trabajo, cada hilo hace su trabajo y al terminar le informa; cuando todos han terminado entonces continua el resto de lógica de negocio. Como pueden leer, es justo lo que estamos necesitando.

Implementación Mejorada

El ajuste en nuestro código se divide en dos partes; primero debemos definir el CountDownLatch. Notese como indicamos la cantidad de threads a esperar y luego de los starts invocamos el método await.

  // supongamos que THREAD_NUMBER es el número de threads a usar
  CountDownLatch latch = new CountDownLatch(THREAD_NUMBER);

  for (int i = 0; i < THREAD_NUMBER; i++) {
       // subLista es una función que nos genera un subset del payload
       // separado para cada Thread. Esto para que cada uno trabaje
       // con sus datos en  paralelo
       new Thread(new ValidatorThread(i, latch,
           subLista(i, payload))).start();
   }
   ...
   latch.await();
   ....

La segunda parte tiene que ver en nuestra implementación del Runnable, en donde debemos invocar el countDown al terminar el procesamiento en ese hilo.

private class ValidatorThread implements Runnable {

  private final CountDownLatch latch;
  private final int threadNumber;
  private final List<MyData> data;
  public ValidatorThread(int i, CountDownLatch latch,
      List<MyData> data) {
     this.threadNumber = i;
     this.data = data;
     this.latch = latch;
  }
  ...
  @Override
  public void run() {
    // ejecutamos nuestra lógica de negocio con base en los
    // datos que tenemos

    latch.countDown();
  }

Con este ajuste; al invocar nuestro API el lambda emplea varios Threads los cuales paralelizan el procesamiento que necesitamos; y como resultado final se logró una mejora sustancial en los tiempos de respuesta del Lambda; llegando a ser de 1 o 2 segundos con los payloads más extensos que se soportan.

Además, este ajuste beneficia a los payloads normales al responder en menor tiempo y por tanto ser más económico por ejecución del API.

Por supuesto, siempre debemos hacer el ejercicio de determinar la mejor selección de memoria para nuestro Lambda a fin de tener la mejor mezcla de precio/rendimiento.

Conclusión

El uso de multithreading en AWS Lambda puede ser sumamente útil cuando podemos paralelizar el trabajo que se debe realizar. Y esto aplica no solo a los Lambda, sino a cualquier otro componente que desarrollemos y que pueda beneficiarse de esto.

Por supuesto, usar múltiples hilos conlleva sus retos y un manejo adecuado en su gestión, caso contrario podríamos llegar a condiciones de competencia, errores en los resultados que esperamos, deadlocks, entre muchos otros elementos desagradables.

Pero con un uso responsable de ellos; las mejores pueden ser sustanciales y además mejorando la economía de las cargas de trabajo nativas que desplegamos en AWS.

Espero que este artículo les haya sido de utilidad.