Multithreading en Node.js

Desde el lanzamiento de la version 10.5.0 de Node.js se puso a disposición el módulo worker_threads y no es hasta la version 12 LTS que el mismo se encontraba estable.

Pero para hablar de Worker Threads es importante antes entender como se encuentra estructurado Node.js.

Como funciona Node.js

Cuando se corre un proceso en Node.js, lo que se corre es:

• Un proceso
• Un hilo
• Un loop de eventos
• Una instancia de motor de JS
• Una instancia de Node.js

Un proceso: un objeto global que puede ser accedido desde cualquier punto dentro del sistema y contiene información sobre que esta siendo ejecutado en un momento específico.

Un hilo: siendo una ejecución de un solo hilo significa que solo un set de instrucciones es ejecutado en un momento específico en un determinado proceso.

Un loop de eventos: Quizas el aspecto mas importante a entender de Node. Permite la ejecución asincrónica y contar con operaciones de I/O no bloqueantes - a pesar de que JS corre en un solo hilo - por medio del aprovisionamiento de operaciones al kernel del sistema cuando sea posible mediante callbacks, promises y async/await.

Una instancia de JS: un solo programa que ejecuta código JavaScript

Una instancia Node.js: un solo programa que ejecuta el código Node.js

En otras palabras, Node corre en un único hilo y solo existe un proceso dentro del loop de eventos. Esto beneficia al desarrollador no teniendo que preocuparse por problemas de concurrencia.

Pero como con todas las cosas esto tiene un aspecto negativo: si se requiere ejecutar procesos largos de uso intensivo de CPU, esta ejecucion puede bloquear otros procesos. Una funcion se considera bloqueante si el loop principal de eventos debe esperar a que finalice antes de ejecutar el próximo comando.

Una funcion "no-bloqueante" permite que el loop de eventos continue la ejecución ni bien comience alertando al loop principal cuando haya finalizado mediante una llamada a un "callback".

Entonces corre en un hilo?

La respuesta es si, podemos ejecutar código en paralelo pero no se crean nuevos hilos de ejecución y no podemos sincronizarlos. Tanto la VM como el sistema operativo ejecutan las operación de I/O en paralelo por nosotros y cuando es momento de ejecutar nuevamente el código JS, la parte de JavaScript es la parte que se encuentra ejecutando dentro de un mismo hilo.

En otras palabras, todo corre en paralelo menos nuestro código JavaScript.

Esto es perfecto si todo lo que realiza nuestro código son operación de I/O. Un código que consista en bloques pequeños sincrónicos que ejecutan de manera veloz pasando datos a archivos y streams, pero que pasa cuando se require el uso intensivo del CPU?

Tareas de uso intensivo de CPU

Que pasaría si necesitamos realizar una tarea que requiera algun tipo de calculo complejo en memoria sobre set de datos grandes? En este caso tendriamos un gran bloque sincronico de codigo que bloquearia el resto del codigo.

Supongamos que el calculo anteriormente mencionado tarda 10 segundos. Si este calculo se ejecuta en un servidor web, significaría que todo request al servidor debería esperar 10s a que finalice el cálculo antes de ser atendido.

Se podria agregar un nuevo módulo al core de Node.js? Por supuesto, pero cambiaría la naturaleza del lenguaje en si. Por ejemplo, en un lenguaje que soporta el multi-threading como Java existeng palabras claves como "synchronized" que permiten que múltiples hilos se sincronicen, pero no es el caso de JavaScript.

La solución

Tomando como referencia la descripción de como se encuentra estructurado Node.js que realizamos al principio, se introduce el concepto de Worker-Threads que nos permite pasar de:

• Un proceso
• Un hilo
• Un loop de eventos
• Una instancia de JS
• Una instancia de Node.js

A...

• Un proceso
• Multiples hilos
• Un loop de eventos por hilo
• Una instancia JS por hilo
• Una instancia Node.js por hilo

El módulo worker_thread nos permite el uso de threads que ejecutan código JavScript en paralelo. La forma de accederlo es la siguiente:

const worker = require('worker_threads');

Lo idea es contar con varias instancias Node.js dentro del mismo proceso. Con Worker threads, un hilo puede finalizar su ejecución en cualquier momento y no necesariamente significa el fín de ejecución de su proceso padre. Pero no es una buena practica que los recursos alocados por un worker queden dando vueltas cuando el worker finaliza la ejecución, esto conlleva a memory leaks. Lo que buscamos es embeber una instancia Node.js dentro de otra, crear un nuevo hilo dentro de un proceso y luego crear una nueva instancia Node.js dentro de este hilo.

Algunas características especiales del módulo:

• Existencia de ArrayBuffers para transferir memoria de un thread a otro
• SharedArrayBuffers accesibles desde cualquier thread. Permite compartir áreas de memoria entre hilos.
• Atomics disponible. Permite la ejecución concurrente y eficiente de un mismo proceso.
• MessagePort, utilizado para la comunicación entre diferentes hilos. Puede ser utilizado para el pasaje de datos estructurados.
• MessageChannel, permite la comunicacion bi-direccional y asincrónica entre diferentes hilos.
• WorkerData es utilizado para pasar un set de datos iniciales al worker thread.

Ejemplo Práctico

Para comenzar a utilizar worker threads se debe importar el módulo worker_threads. Comencemos por crear una función que nos ayude a que estos worker_threads "aparezcan" y luego veamos las propiedades que aquí definimos.

type WorkerCallback = (err: any, result?: any) => any;
export function runWorker(path: string, cb: WorkerCallback, workerData: object | null = null) {
 const worker = new Worker(path, { workerData });
 worker.on('message', cb.bind(null, null));
 worker.on('error', cb);
 worker.on('exit', (exitCode) => {
   if (exitCode === 0) {
     return null;
   }
   return cb(new Error(`Worker has stopped with code ${exitCode}`));
 });
 return worker;
}

A fín de crear un worker, tenemos que crear una instancia de la clase Worker. Los argumentos que recibe esta clase son, en primer lugar, la ruta al archivo que contiene el código del worker y en segundo lugar un objeto que contiene la propiedad workerData, esta es la información que queremos que el thread tenga disponible cuando comience su ejecución.

Como se puede observar en el ejemplo anterior, la comunicación es basada en eventos, lo que significa que definimos listeners que ejecuten código una vez recibido un determinado evento por parte del worker thread.

A continuación vemos una lista de los eventos mas comunes:

worker.on('error', (error) => {});

El evento  error es emitido por el worker thread siempre que exista una excepción no atrapada dentro del código. En este caso, el worker thread finaliza y el error se disponibilza como el primer argumento del callback provisto.

worker.on('exit', (exitCode) => {});

El evento exit se emite siempre que el worker llegue al final de ejecución. Si dentro del código se llama a process.exit(), donde se proveera como argumento a la funcion de callback un exitCode. Si el worker se finaliza mediante worker.terminate(), el código sera 1.

worker.on('online', () => {});

online se emite siempre que se termine la lectura de todo el código fuente referenciado y comienza la ejecución.

worker.on('message', (data) => {});

El evento message se emite cuando un worker envía datos a su proceso padre.

Intercambiando datos entre hilos

A fin de enviar datos de un thread a otro se utiliza el método port.postMessage() con la siguiente firma

port.postMessage(data[, transferList])

El objeto port puede ser tanto parentPort o una instancia de MessagePort como se verá mas adelante.

El primer argumento es un objeto que es copiado en el otro thread. Puede contener cualquier cosa que sea soportada por los algoritmos de copia estructurados.

El algoritmo no copia funciones, errores, descriptor de propiedades o prototipos encadenados. Al soportar la copia de Arreglos tipados, el algoritmo permite compartir áreas de memoria entre hilos.

El segundo argumento, sólo puede contener objetos ArrayBuffer o MessagePort. Una vez enviados al hilo receptor no puede ser utilizados por el hilo que los envía; el área de memoria es transferida al hilo receptor.

Creando canales de comunicacion

La comunicación entre threads se realiza a través de puertos que no son otra cosa mas que instancias de la clase MessagePort y posibilita la comunicación basada en eventos.

Existen dos formas de utilizar los puertos para la comunicación entre threads siendo la primera la mas sencilla de los dos. Dentro del código fuente del worker se importa un objeto llamado parentPort del módulo worker_threads y se utiliza el metodo postMessage() para enviar mensajes al hilo padre.

Por ejemplo:

import { parentPort } from 'worker_threads';
const data = {
// ...
};
parentPort.postMessage(data);

El objeto parentPort es una instancia de MessagePort que Node.js crea para nosotros para permitir la comunicación con el hilo padre.

La segunda forma de comunicación entre threads es crear un MessageChannel por nuestra cuenta enviándoselo al worker thread como se ve a continuación:

import path from 'path';
import { Worker, MessageChannel } from 'worker_threads';
const worker = new Worker(path.join(__dirname, 'worker.js'));
const { port1, port2 } = new MessageChannel();
port1.on('message', (message) => {
 console.log('message from worker:', message);
});
worker.postMessage({ port: port2 }, [port2]);

Luego de haber creado port1 y port2, se establece un listener sobre port1 y se envia port2 al worker thread. Se debe incluirlo dentro del transferList a fin de que sea transferido al worker receptor.

Y luego dentro del worker thread:

import { parentPort, MessagePort } from 'worker_threads';
parentPort.on('message', (data) => {
 const { port }: { port: MessagePort } = data;
 port.postMessage('heres your message!');
});

De esta manera podemos usar el puerto enviado por el hilo padre.

Conclusiones

Algunas personas podrán argumentar que módulos como cluster o child_process permitían el usode threads desde hace algún tiempo ya.

El módulo cluster permite la creación de múltiples instancias Node.js dentro de un proceso encargado del enrutamiento de los requests hacia las instancias. Ejecutar una aplicación en cluster nos permite de manera efectiva multiplicar el throughput de un servidor, sin embargo, no se puede instanciar nuevos hilos dentro del módulo cluster.

Por lo general las personas tienden a utilizar herramientas como PM2 para correr sus aplicaciones en cluster en vez de realizarlo de forma manual.

El módulo child_process puede instanciar cualquier ejecutable independientemente de que sea o no JavaScript. Bastante similar a worker_threads pero carece de algunos features importantes.

Especicificamente, thread workers son mas "livianos" en términos de ejecución y comparten el mismo identificador de proceso que el proceso padre. Pueden compartir areas de memoria con su proceso padre, evitando serializar grandes set de datos y enviarlos de forma bidireccional de manera mas efectiva.