Accélérez les applications JavaScript avec des pipelines simultanés | de Kartal Kaan Bozdogan | décembre 2022

Comment nous avons obtenu une accélération de 35 % sur notre application

Les pipelines de données, comme le célèbre pipeline ETL, ne sont guère étrangers à la plupart des développeurs back-end.

Je me suis récemment retrouvé à travailler avec un tel pipeline dans le backend Node.js de mon équipe. Je travaillais sur un point de terminaison prenant un certain nombre d’objets, les validant et les insérant dans une base de données :

for (var obj of objects) {
await validate(obj);
await insert(obj);
}

Ce modèle était dans la base de code depuis un certain temps. Auparavant, il faisait le travail, mais n’était pas particulièrement efficace, car chaque objet nécessitait un aller-retour vers le backend :

Après avoir remarqué qu’une quantité importante de temps humain était gaspillée à attendre l’opération, j’ai optimisé le point de terminaison de sorte qu’il utilise insertions de base de données en masse. Cela réduit le nombre d’allers-retours vers la base de données à 1 :

const bulk = new Bulk();
for (var obj of objects) {
await validate(obj);
bulk.insert(obj);
}
await bulk.execute();

Voici à quoi ressemble la chronologie après le changement :

Les résultats ont été excellents : les performances du terminal ont atteint des niveaux satisfaisants.

Mais il existe une autre opportunité d’optimisation : remarquez comment un objet en cours de validation est complètement indépendant d’un autre en cours d’insertion, mais toutes les insertions de base de données se produisent après toutes les validations d’objets. Et si nous chevauchions la validation d’objet et l’insertion de base de données, comme ceci :

Mais comment implémenter un tel comportement en JavaScript ? Une façon pourrait être de laisser tomber le dernier await tout à fait:

for (var batchStart=0; batchStart<objects.length; batchStart+=BATCH_SIZE) {
const bulk = new Bulk();
for (var i=batchStart; i<batchStart+BATCH_SIZE && i<objects.length; i++) {
await validate(objects[i]);
bulk.insert(objects[i]);
}
bulk.execute();
}

Mais alors, bien sûr, il n’y a pas de limite au nombre de requêtes de base de données simultanées en cours, surtout si la validation prend peu de temps par rapport à l’insertion de la base de données et que nous travaillons avec un grand nombre d’objets, pas pour mentionner la gestion des exceptions asynchrones.

Pour résoudre ce problème, nous pourrions utiliser concurrent-pipeline, un package NPM créé par moi. Nous commençons par importer le package et créer un nouveau Pipeline :

const Pipeline = require('concurrent-pipeline');
const ppl = new Pipeline(10);

Le 10 signifie « au plus 10 flux en cours d’exécution simultanément ».

Ensuite, nous enveloppons la boucle for interne avec pipelinedpermettant à la boucle for externe de continuer à progresser sans avoir à attendre chaque lot :

for (var batchStart=0; batchStart<objects.length; batchStart+=BATCH_SIZE) {
await ppl.pipelined(async (stage, batchStart) => {
const bulk = new Bulk();
for (var i=batchStart; i<batchStarts+BATCH_SIZE && i<objects.length; i++) {
await validate(objects[i]);
bulk.insert(objects[i]);
}
await bulk.execute();
})(batchStart);
}

Nous pouvons appliquer différentes limites de simultanéité aux différentes étapes de notre code. Supposons que nous souhaitions au maximum 5 validations simultanées, mais que l’étape d’insertion de la base de données soit limitée à 3 opérations simultanées :

for (var batchStart=0; batchStart<objects.length; batchStart+=BATCH_SIZE) {
await ppl.pipelined(async (stage, batchStart) => {
const bulk = new Bulk();
await stage('validation', 5);
for (var i=batchStart; i<batchStarts+BATCH_SIZE && i<objects.length; i++) {
await validate(objects[i]);
bulk.insert(objects[i]);
}
await stage('database', 3);
await bulk.execute();
})(batchStart);
}

Le await stage Les instructions empêchent les flux de progresser vers des étapes qui ont déjà atteint leur limite de simultanéité, jusqu’à ce qu’une place se libère.

Enfin, nous attendons que tous les flux se terminent :

await ppl.finish();

Cet appel lèvera également toutes les exceptions qui n’ont pas été gérées par les flux.

En utilisant concurrent-pipelinepour chevaucher les validations avec les opérations de base de données et maintenir plusieurs opérations de base de données en cours, j’ai atteint une accélération supplémentaire de 35 % sur notre application.

Vous pouvez trouver plus de détails sur concurrent-pipeline sur son dépôt.