JavaScript es un lenguaje de programación de un solo subproceso, lo que significa que solo puede ejecutar una tarea a la vez. Esto puede convertirse en un problema cuando se ejecutan tareas pesadas y de larga duración, como el procesamiento de imágenes o el análisis de datos de gran tamaño, ya que puede hacer que el subproceso principal se congele y que la aplicación deje de responder.

Para resolver este problema, el ecosistema de JavaScript ha desarrollado una técnica llamada subprocesamiento. La creación de subprocesos le permite ejecutar tareas en un subproceso separado, para que no bloqueen el subproceso principal y afecten el rendimiento de su aplicación.

En este artículo, presentaremos un paquete npm llamado función de paralelismo, que le permite ejecutar funciones de JavaScript en un subproceso diferente. El paquete usa la API Worker, que funciona tanto para navegadores como para Node.js.

¿Qué es la función de paralelizador?

Parallelizer-function es un paquete npm que le permite ejecutar funciones de JavaScript en subprocesos separados, utilizando la API Worker. Funciona tanto para navegadores como para NodeJs. Según el entorno de ejecución, utiliza la biblioteca incorporada de Nodejs "worker_threads" o la clase window.Worker predeterminada en el caso de un entorno de navegador.

Instalación

Para instalar la función de paralelizador, puede usar npm o yarn:

npm i parallelizer-function --save

yarn add parallelizer-function

El paquete tiene dos partes principales principales: la función workPromise y la clase Pool. La función workerPromise le permite ejecutar una función en un subproceso separado, y la clase Pool implementa un grupo de subprocesos que ejecuta cada tarea con un número máximo predefinido de subprocesos o trabajadores. Implementa una cola asíncrona para ejecutar la función una vez que un trabajador disponible ha terminado su trabajo actual.

workerPromisefunción

La workerPromisefunción es una función simple que toma dos argumentos, la función a ejecutar y una matriz de argumentos para esa función. Devuelve una promesa que resuelve el resultado de la función o rechaza con un error.

const { workerPromise } = require("parallelizer-function");

const longRunningTask = (n) => {
  let result = 0;
  for (let i = 0; i < n; i++) {
    result += i;
  }
  return result;
}

async function main() {
    // workerPromise(fn: (...params: any[]) => any, args: any[] = []): Promise<any>
    workerPromise(longRunningTask,[1_000_000])
        .then((res)=>{console.log(res)})
        .catch((error)=>{console.error(error)})

    /// or using try-catch
    try{
        let res = await workerPromise(longRunningTask,[1_000_000]);
        console.log(res);
    }catch(error){
        console.error(error)
    }
}

Poolclase: limitar el número de subprocesos simultáneos

La Poolclase proporciona una forma de limitar la cantidad de subprocesos simultáneos, de modo que su aplicación no sobrecargue el sistema. Esto es útil para ejecutar tareas de larga duración, como el procesamiento de imágenes, y ejecutar una gran cantidad de tareas al mismo tiempo.

const { Pool, pool } = require('parallelizer-function');

// It you want to maintain the global state of the pool use the object pool, but you can define a new Object: const newPool =  new Pool(2)

pool.setMaxWorkers(2); // Set the maximum number of worker threads for the global object pool

const longRunningTask = (n) => {
  let result = 0;
  for (let i = 0; i < n; i++) {
    result += i;
  }
  return result;
};

const heavyImageProcessing = (imageData: ImageData) => {
  for (let i = 0; i < imageData.data.length; i += 4) {
    // This is a heavy operation that will block the main thread
    imageData.data[i] = imageData.data[i] * 2;
    imageData.data[i + 1] = imageData.data[i + 1] * 2;
    imageData.data[i + 2] = imageData.data[i + 2] * 2;
  }
  return imageData;
};

async function main() {
  try {
   //pool.exec(fn: (...params: any[]) => any, args: any[]): Promise<any>
    let res = await pool.exec(longRunningTask, [1_000_000]);
    console.log(res);
  } catch (error) {
    console.error(error);
  }

  /// You can use Promise.all to run several tasks

  try {
    Promise.all([
      pool.exec(longRunningTask, [1_000_000]),
      pool.exec(longRunningTask, [2_000_000]),
      pool.exec(heavyImageProcessing, [imageData]),
    ]).then((res) => {
      console.log(res);
    });
    console.log(
      'This function will be executed very fast, the above heavy tasks will not block the main thread'
    );
  } catch (error) {
    console.error(error);
  }
}

Lo siguiente que hacemos es establecer la cantidad máxima de subprocesos de trabajo para el objeto global poolusando el setMaxWorkers(2)método. Esto significa que solo estamos permitiendo que 2 subprocesos de trabajo se ejecuten simultáneamente.

Luego definimos dos funciones, longRunningTasky heavyImageProcessing, que simulan tareas de ejecución prolongada. La longRunningTaskfunción simplemente realiza un cálculo para un número determinado de iteraciones, mientras que la heavyImageProcessingfunción realiza una operación pesada en una imagen.

En la mainfunción, estamos usando el pool.exec()método para ejecutar nuestra longRunningTaskfunción con el argumento [1_000_000].

Limitaciones

1. La función de trabajo debe poder ser encadenada o clonada (por ejemplo, no puede ser un método de clase)
2. Todas las bibliotecas o paquetes que utiliza la función para realizar su tarea deben importarse dentro de la función. Esto se debe a que los trabajadores se ejecutan en otro contexto global que es diferente de la ventana actual. La función se aislará como si estuviera en un script separado.
3. Puede ejecutar cualquier código que desee dentro del subproceso de trabajo, con algunas excepciones. Por ejemplo, no puede manipular directamente el DOM desde dentro de un trabajador, o usar algunos métodos y propiedades predeterminados del objeto de la ventana

import { pool } from "parallelizer-function";
import path from "path";

//////////// main script
try{
    let pathToFile = path.resolve("../docs/sample-name.txt");

    let res:Array<string> = await pool.exec(
        async (pathFile)=>{
            const fs = require('fs');
            let files = fs.readFileSync(pathFile, { encoding: 'utf-8' });
            return files.split('\n');
        }, [pathToFile]);

    console.log("Names: ", res)
}catch(error){
    console.error(error)
}

Estos son algunos ejemplos de cómo puede usar este paquete en su aplicación:

1. Ejecutar una tarea de ejecución prolongada, como el procesamiento de imágenes, en un subproceso separado para evitar que bloquee el subproceso principal.
2. Ejecutar una gran cantidad de tareas al mismo tiempo, pero limitando la cantidad de subprocesos para evitar que el sistema se sobrecargue.
3. Ejecutar una tarea en un hilo separado y usar una función de devolución de llamada para recuperar el resultado

import { Pool } from "parallelizer-function";

let pool = new Pool(4); // Here we instanciate another instance of Worker Pool and we set a pool of 4 workers

function isPrimeThisNumber(n){
    // This function takes an integer and returns whether it is a prime number or not. Complexity O(n^1/2)
    for(let i=2;i*i<=n;i++){
        if(n%i == 0) return false;
    }
    return true
}

function TripleSum(arr=[]){
    // This function return all the distinc triplet i,j,k i<j<k,
    // where arr[i] + arr[j] + arr[k] sum up to 0. Complexity O(n^2)
    let visited = new Set()
    let sol = []
    arr = arr.sort()
    for(let i =0;i<arr.length;i++){
        let target = -arr[i]
        let isSeen = new Set()
        for(let j=i+1;j<arr.length;j++){
            if(isSeen.has(target - arr[j])){
                let key = `${arr[i]},${arr[j]},${target - arr[j]}`
                if(!visited.has(key))
                    sol.push([arr[i],arr[j],target - arr[j]])
                visited.add(key)
            }else{
                isSeen.add(arr[j])
            }

        }
    }
    return sol
}

function simulateLongTask(delayS = 10){
    // This function simulate a task that will take <delayS> seconds to finish
    let now = Date.now();
    let iter = 0;
    let MAX_DELAY = delayS * 1000; // 10 seconds 100000 milliseconds
    while((Date.now() - now) < MAX_DELAY ){
        iter++;
    }
    return iter;
}

someBTNEl.addEventListener("click",()=>{
    console.log(isPrimeThisNumber(352684978))
    /// This will block the main thread
})
// Also this

someBTNEl.addEventListener("click",()=>{
    let promiseFn = new Promise((resolve)=>{
        resolve(isPrimeThisNumber(352684978));
    })
    promiseFn.then(console.log);
    /// This also will block the main thread.
    // Wrapping normal code in a Promise does not guarantee that the main thread will not be blocked.
})

someBTNEl.addEventListener("click",async ()=>{
    try{
        let res = await pool.exec(isPrimeThisNumber,[352684978]);
        console.log(res);
    // This will not block the main thread of JS, it will run "isPrimeThisNumber"
    // in a separate thread using Worker class.

    }catch(error){

    }

})
// You can do all computation in once
someBTNEl.addEventListener("click",async ()=>{
    Promise.all([
        pool.exec(isPrimeThisNumber,[352684978]),
        pool.exec(simulateLongTask,[10])
    ]).then((isPrime,sum] )=>{
        /// do more stuff
    })
})

// Or in an endpoind for the computation of the functions
let functions = {
    TripleSum,
    isPrimeThisNumber,
    simulateLongTask
}

app.post("/compute/:fn",(req,res)=>{
    let fn = req.params.fn;
    if(!fn || !(fn in functions)){
        return res.status(401).json({msg:"Not found the function"});
    }
    try{
        let res = await pool.exec(functions[fn],req.body?.args || []);
        // This will not block the main thread of JS, it will run
        return res.status(200).json({ error:false,msg:"OK",data:res });
    }catch(e){
        return res.status(400).json({ error:true,msg:e.message });
    }
})

Ejemplo de procesamiento de imagen

En el siguiente código recortado, la processImagefunción toma un imagetipo de HTMLImageElementy crea un elemento de lienzo, dibuja la imagen en él y luego obtiene el ImageDatadel lienzo. Luego ejecuta la heavyImageProcessingfunción en la imageDataque es una operación pesada que bloquearía el hilo principal y luego vuelve a colocar el procesamiento imageDataen el lienzo y devuelve la URL de datos del lienzo.

La processImageInWorkerfunción es muy similar a la processImagefunción, pero en lugar de ejecutar la heavyImageProcessingfunción directamente, ejecuta la función usando el pool.execmétodo, pasando la heavyImageProcessingfunción y imageDatacomo argumentos. Esto ejecutará la heavyImageProcessingfunción en un subproceso separado utilizando el poolobjeto y no bloqueará el subproceso principal. Luego imageDatavuelve a colocar el procesamiento en el lienzo y devuelve la URL de datos del lienzo.

import { pool } from "parallelizer-function";

const heavyImageProcessing = (imageData: ImageData) => {
  for (let i = 0; i < imageData.data.length; i += 4) {
    // This is a heavy operation that will block the main thread
    imageData.data[i] = imageData.data[i] * 2;
    imageData.data[i + 1] = imageData.data[i + 1] * 2;
    imageData.data[i + 2] = imageData.data[i + 2] * 2;
  }
  return imageData;
};

const processImage = async (image: HTMLImageElement) => {
  const canvas = document.createElement('canvas');
  canvas.width = image.width;
  canvas.height = image.height;
  const ctx = canvas.getContext('2d');
  ctx.drawImage(image, 0, 0);
  const imageData = ctx.getImageData(0, 0, canvas.width, canvas.height);

  // This will block the main thread
  const processedImageData = heavyImageProcessing(imageData);
  ctx.putImageData(processedImageData, 0, 0);

  return canvas.toDataURL();
};

const processImageInWorker = async (image: HTMLImageElement) => {
  const canvas = document.createElement('canvas');
  canvas.width = image.width;
  canvas.height = image.height;
  const ctx = canvas.getContext('2d');
  ctx.drawImage(image, 0, 0);
  const imageData = ctx.getImageData(0, 0, canvas.width, canvas.height);

  // This will not block the main thread
  const processedImageData = await pool.exec(heavyImageProcessing, [imageData]);
  ctx.putImageData(processedImageData, 0, 0);

  return canvas.toDataURL();
};

En este ejemplo en la ruta “/” de la aplicación se está devolviendo el resultado de computar un conjunto de tareas largas. Si no pasamos el parámetro de consulta, usingPool=truela ejecución se ejecutará en el hilo principal de NodeJs.

En la Figura 2 anterior, cuando alcanzamos el punto final “/”, el tiempo que tarda en finalizar la ejecución de mainHeavyTaskes de 2347 s. Pero dado que estas tareas se ejecutan en el hilo principal, todas las solicitudes entrantes se bloquean esperando su respuesta hasta que se libera el hilo principal.

En la Figura 3 anterior, cuando llegamos al punto final “/”, pero ahora utilizando el grupo de 4 trabajadores, el tiempo que tarda en terminar la ejecución de mainHeavyTaskes de 1,5 s (950 ms menos que el anterior). Pero como estas tareas se ejecutan en diferentes subprocesos, todas las solicitudes entrantes se ejecutarán con normalidad y su ejecución no se verá afectada por la ejecución de mainHeavyTask.

Aquí otro ejemplo usando reaccionar

Conclusión

El grupo de subprocesos le permite ejecutar sus funciones en subprocesos separados, para que no bloqueen el subproceso principal. También le permite limitar la cantidad de subprocesos concurrentes, para que su aplicación no sobrecargue el sistema. Esto es útil para ejecutar tareas de larga duración, como el procesamiento de imágenes, y ejecutar una gran cantidad de tareas al mismo tiempo.

Nota

Este paquete ha sido probado en Node.js v14.x v16.x, v18.x y en navegadores como Mozilla y Chrome usando Vanilla JavaScript, React y Angular

Autor

autor: José Alejandro Concepción Álvarez

repo: función de paralelizador