Exécution parallèle de tâches asynchrones

Au Parc Jurassique, gérer simultanément de nombreux systèmes et surveiller des dizaines de dinosaures nécessite une approche efficace pour exécuter de nombreuses tâches en même temps. Dans cet exercice, nous nous concentrerons sur les techniques d'exécution parallèle de tâches asynchrones en JavaScript, qui nous permettront d'améliorer les performances de nos systèmes de gestion du parc.

Pourquoi l'exécution parallèle est-elle importante ?

Imagine que nous devons vérifier l'état des 50 enclos de dinosaures avant d'ouvrir le parc. Si chaque vérification prend environ 1 seconde et que nous les exécutons séquentiellement (l'une après l'autre), tout le processus prendra au moins 50 secondes - bien trop long avant de laisser entrer les visiteurs !

En exécutant ces tâches en parallèle, nous pouvons réduire considérablement le temps total nécessaire à toutes les opérations, en le limitant à la durée de la seule opération la plus longue (plus un peu de surcoût).

Promise.all() - l'outil de base de l'exécution parallèle

Comme nous l'avons vu dans les leçons précédentes, Promise.all() est l'outil fondamental pour exécuter de nombreuses tâches asynchrones en parallèle :

1async function checkAllEnclosureStatuses() {
2  const enclosureList = await fetchEnclosureList();
3  console.log(`Vérification de l'état de ${enclosureList.length} enclos...`);
4
5  const startTime = Date.now();
6
7  // Lancer la vérification de tous les enclos en parallèle
8  const results = await Promise.all(
9    enclosureList.map(enclosure => checkEnclosureStatus(enclosure.id))
10  );
11
12  const executionTime = (Date.now() - startTime) / 1000;
13  console.log(`La vérification de tous les enclos a pris ${executionTime.toFixed(2)} secondes`);
14
15  // Analyser les résultats
16  const secureEnclosures = results.filter(w => w.status === 'secure');
17  const problematicEnclosures = results.filter(w => w.status !== 'secure');
18
19  console.log(`Enclos sécurisés : ${secureEnclosures.length}`);
20  console.log(`Enclos problématiques : ${problematicEnclosures.length}`);
21
22  if (problematicEnclosures.length > 0) {
23    console.warn("Problèmes détectés dans les enclos suivants :");
24    problematicEnclosures.forEach(w => {
25      console.warn(`- Enclos n°${w.id} : ${w.status} (${w.message})`);
26    });
27  }
28
29  return {
30    total: results.length,
31    secure: secureEnclosures.length,
32    problematic: problematicEnclosures,
33    executionTime
34  };
35}

Dans l'exemple ci-dessus, nous utilisons Promise.all() pour vérifier l'état de tous les enclos en parallèle. Au lieu d'attendre la fin de chaque vérification avant de démarrer la suivante, toutes les vérifications sont lancées simultanément et nous attendons qu'elles se terminent toutes.

Défis de l'exécution parallèle

L'exécution parallèle de tâches asynchrones apporte de nombreux avantages, mais comporte aussi certains défis :

1. Gestion des ressources

Lancer trop d'opérations en parallèle peut surcharger les ressources système disponibles (mémoire, connexions réseau, etc.).

2. Gestion des erreurs

Souviens-toi que Promise.all() échouera si une seule promise est rejetée. Avec un grand nombre d'opérations parallèles, une seule erreur peut faire échouer tout le processus.

3. Trop de parallélisme

Exécuter toutes les tâches en parallèle n'est pas toujours la meilleure solution. Parfois, il faut contrôler le nombre maximal d'opérations parallèles.

Contrôler la concurrence - pas trop à la fois

Exécuter trop d'opérations simultanément peut surcharger le serveur ou l'API. Il nous faut un moyen d'exécuter les tâches en parallèle, mais avec une limite sur le nombre exécuté en même temps :

1async function runWithConcurrencyLimit(tasks, limit) {
2  const results = [];
3  const executing = [];
4
5  for (const task of tasks) {
6    const promise = Promise.resolve().then(() => task());
7    results.push(promise);
8
9    if (limit <= tasks.length) {
10      const executor = promise.then(() => executing.splice(executing.indexOf(executor), 1));
11      executing.push(executor);
12
13      if (executing.length >= limit) {
14        await Promise.race(executing);
15      }
16    }
17  }
18
19  return Promise.all(results);
20}
21
22// Vérifier tous les enclos mais 5 au maximum à la fois
23async function checkAllPaddocksLimited() {
24  const paddockIds = Array.from({ length: 50 }, (_, i) => i + 1);
25
26  const checkTasks = paddockIds.map(id => () => checkSinglePaddock(id));
27
28  console.time("parallel-limited");
29  const results = await runWithConcurrencyLimit(checkTasks, 5);
30  console.timeEnd("parallel-limited");
31
32  return results;
33}
34
35async function checkSinglePaddock(paddockId) {
36  await new Promise(res => setTimeout(res, 200 + Math.random() * 300));
37  return { paddockId, status: Math.random() > 0.1 ? "secure" : "alert", checkedAt: new Date().toISOString() };
38}

Bibliothèques pour gérer le parallelisme

Plusieurs bibliothèques populaires facilitent la gestion de l'exécution parallèle des tâches :

p-limit

La bibliothèque p-limit permet de limiter facilement les opérations parallèles :

1const pLimit = require('p-limit');
2
3async function monitorAllDinosaurs() {
4  const dinosaurList = await fetchDinosaurList();
5
6  // Créer une instance de limiteur avec 5 opérations parallèles au maximum
7  const limit = pLimit(5);
8
9  // Créer un tableau de promises avec parallélisme limité
10  const promises = dinosaurList.map(dino => {
11    return limit(() => monitorDinosaur(dino.id));
12  });
13
14  // Attendre la fin de toutes les opérations
15  const results = await Promise.all(promises);
16
17  return results;
18}

Traitement parallèle avec gestion des erreurs

Comme nous l'avons mentionné plus tôt, Promise.all() échouera si une seule promise est rejetée. Dans certains cas, nous voulons continuer le traitement même si certaines opérations échouent.

En ECMAScript 2020, Promise.allSettled() a été introduit, ce qui résout ce problème :

1async function runDiagnosticsForAllSystems() {
2  const systems = await fetchAllSystems();
3
4  console.log(`Lancement des diagnostics pour ${systems.length} systèmes...`);
5
6  // Lancer les diagnostics pour tous les systèmes en parallèle et attendre qu'ils se terminent tous
7  const results = await Promise.allSettled(
8    systems.map(async (system) => {
9      try {
10        console.log(`Diagnostics pour le système ${system.name}...`);
11        const result = await systemDiagnostics(system.id);
12        console.log(`Diagnostics du système ${system.name} terminés`);
13        return { systemId: system.id, status: 'success', result: result };
14      } catch (error) {
15        console.error(`Erreur de diagnostics pour le système ${system.name} :`, error);
16        return { systemId: system.id, status: 'error', error: error.message };
17      }
18    })
19  );
20
21  // Analyser les résultats
22  const successful = results.filter(w => w.status === 'fulfilled' && w.value.status === 'success');
23  const failed = results.filter(w => w.status === 'rejected' || (w.status === 'fulfilled' && w.value.status === 'error'));
24
25  console.log(`Diagnostics terminés : ${successful.length} réussis, ${failed.length} échoués`);
26
27  return {
28    total: results.length,
29    successful: successful.length,
30    failed: failed.length,
31    results
32  };
33}

Traitement par lots (batch)

Parfois, il vaut mieux traiter de grandes collections d'éléments par lots :

1async function processDinosaursBatch(dinosaurIds, batchSize = 10) {
2  const results = [];
3
4  for (let i = 0; i < dinosaurIds.length; i += batchSize) {
5    const batch = dinosaurIds.slice(i, i + batchSize);
6    console.log(`Traitement du lot ${Math.floor(i / batchSize) + 1} : dinosaures ${i + 1}-${Math.min(i + batchSize, dinosaurIds.length)}`);
7
8    // Traiter le lot en parallèle
9    const batchResults = await Promise.all(
10      batch.map(id => processSingleDinosaur(id))
11    );
12
13    results.push(...batchResults);
14
15    // Optionnel : petite pause entre les lots pour éviter les limites de débit de l'API
16    if (i + batchSize < dinosaurIds.length) {
17      await new Promise(res => setTimeout(res, 100));
18    }
19  }
20
21  return results;
22}
23
24async function processSingleDinosaur(id) {
25  await new Promise(res => setTimeout(res, 100 + Math.random() * 200));
26  return { id, processed: true, health: Math.floor(70 + Math.random() * 30) };
27}
28
29// Utilisation
30const allDinoIds = Array.from({ length: 100 }, (_, i) => i + 1);
31processDinosaursBatch(allDinoIds, 10).then(results => {
32  console.log(`${results.length} dinosaures traités`);
33  const avgHealth = results.reduce((sum, d) => sum + d.health, 0) / results.length;
34  console.log(`Santé moyenne : ${avgHealth.toFixed(1)}%`);
35});

File de priorité

Certaines opérations sont plus urgentes que d'autres. Une file de priorité garantit que les tâches critiques s'exécutent en premier :

1class PriorityTaskQueue {
2  constructor(concurrency = 3) {
3    this.concurrency = concurrency;
4    this.running = 0;
5    this.queues = { high: [], medium: [], low: [] };
6  }
7
8  async add(task, priority = 'medium') {
9    return new Promise((resolve, reject) => {
10      this.queues[priority].push({ task, resolve, reject });
11      this.run();
12    });
13  }
14
15  run() {
16    while (this.running < this.concurrency) {
17      const next = this.queues.high.shift() ||
18                   this.queues.medium.shift() ||
19                   this.queues.low.shift();
20
21      if (!next) break;
22
23      this.running++;
24      Promise.resolve()
25        .then(() => next.task())
26        .then(result => {
27          next.resolve(result);
28        })
29        .catch(error => {
30          next.reject(error);
31        })
32        .finally(() => {
33          this.running--;
34          this.run();
35        });
36    }
37  }
38}
39
40// Utilisation
41const parkTaskQueue = new PriorityTaskQueue(4);
42
43// Priorité haute - alertes de sécurité
44async function handleSecurityAlert(alertId) {
45  return parkTaskQueue.add(
46    () => processSecurityAlert(alertId),
47    'high'
48  );
49}
50
51// Priorité moyenne - surveillance régulière
52async function scheduledMonitoring(paddockId) {
53  return parkTaskQueue.add(
54    () => checkSinglePaddock(paddockId),
55    'medium'
56  );
57}
58
59// Priorité basse - synchronisation des données
60async function syncHistoricalData(dinoId) {
61  return parkTaskQueue.add(
62    () => syncDinoHistory(dinoId),
63    'low'
64  );
65}
66
67function processSecurityAlert(id) { return new Promise(res => setTimeout(() => res({ alertId: id, handled: true }), 150)); }
68function syncDinoHistory(id) { return new Promise(res => setTimeout(() => res({ dinoId: id, synced: true }), 400)); }

Streaming des résultats

Lorsqu'on traite de nombreux éléments, on peut diffuser (stream) les résultats au fur et à mesure de leur achèvement plutôt que d'attendre qu'ils soient tous terminés :

1async function* processWithStreaming(items, processor) {
2  const promises = items.map((item, index) =>
3    processor(item).then(result => ({ index, result }))
4  );
5
6  // Utiliser un tracker pour produire les résultats dans l'ordre d'achèvement
7  const pending = new Map(promises.map((p, i) => [i, p]));
8
9  while (pending.size > 0) {
10    // Attendre que la prochaine promise se termine
11    const settled = await Promise.race(pending.values());
12    pending.delete(settled.index);
13    yield settled.result;
14  }
15}
16
17// Utilisation - voir les résultats au fur et à mesure de leur arrivée
18async function monitorDinosaursStreaming(dinoIds) {
19  const processor = async (id) => {
20    await new Promise(res => setTimeout(res, 100 + Math.random() * 900));
21    return { id, health: Math.floor(60 + Math.random() * 40), timestamp: Date.now() };
22  };
23
24  let count = 0;
25  for await (const result of processWithStreaming(dinoIds, processor)) {
26    count++;
27    console.log(`[${count}/${dinoIds.length}] Dinosaure ${result.id} : ${result.health}% de santé`);
28  }
29}
30
31monitorDinosaursStreaming([1, 2, 3, 4, 5, 6, 7, 8]);

Worker Threads dans Node.js

Dans Node.js, nous pouvons utiliser le module worker_threads pour effectuer des calculs intensifs dans des threads séparés :

1const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
2
3// Code principal
4if (isMainThread) {
5  async function analyzeDinosaurData() {
6    const data = await fetchLargeDataSets();
7
8    // Diviser les données en parties, une par cœur de CPU
9    const threadCount = require('os').cpus().length;
10    const dataParts = splitData(data, threadCount);
11
12    // Créer un worker pour chaque partie de données
13    const workers = dataParts.map((part, index) => {
14      return new Promise((resolve, reject) => {
15        const worker = new Worker(__filename, {
16          workerData: { data: part, index }
17        });
18
19        worker.on('message', resolve);
20        worker.on('error', reject);
21        worker.on('exit', (code) => {
22          if (code !== 0) {
23            reject(new Error(`Le worker s'est arrêté avec le code de sortie ${code}`));
24          }
25        });
26      });
27    });
28
29    // Attendre la fin de tous les workers
30    const results = await Promise.all(workers);
31
32    // Fusionner les résultats
33    return mergeResults(results);
34  }
35} else {
36  // Code exécuté dans chaque thread worker
37  const { data, index } = workerData;
38
39  // Effectuer des calculs complexes sur les données
40  const results = processData(data);
41
42  // Renvoyer les résultats au thread principal
43  parentPort.postMessage(results);
44}

Web Workers pour les tâches gourmandes en CPU

Pour des calculs lourds qui bloqueraient le thread principal dans le navigateur, utilise les Web Workers :

1// main.js - création d'un Web Worker
2function analyzeDinoPopulationStats(dinosaurData) {
3  return new Promise((resolve, reject) => {
4    const worker = new Worker('dino-stats-worker.js');
5
6    worker.postMessage({ type: 'ANALYZE', data: dinosaurData });
7
8    worker.onmessage = (event) => {
9      if (event.data.type === 'RESULT') {
10        resolve(event.data.result);
11        worker.terminate();
12      }
13    };
14
15    worker.onerror = (error) => {
16      reject(error);
17      worker.terminate();
18    };
19  });
20}
21
22// dino-stats-worker.js - s'exécute dans un thread séparé
23self.onmessage = function(event) {
24  if (event.data.type === 'ANALYZE') {
25    const { data } = event.data;
26
27    // Calcul gourmand en CPU - ne bloque pas le thread principal
28    const stats = {
29      totalWeight: data.reduce((sum, d) => sum + d.weight, 0),
30      avgAge: data.reduce((sum, d) => sum + d.age, 0) / data.length,
31      speciesDistribution: data.reduce((acc, d) => {
32        acc[d.species] = (acc[d.species] || 0) + 1;
33        return acc;
34      }, {}),
35      dangerIndex: data.reduce((sum, d) => sum + d.dangerLevel, 0) / data.length
36    };
37
38    self.postMessage({ type: 'RESULT', result: stats });
39  }
40};

Processeur de tâches adaptatif

Une solution complète qui adapte la concurrence en fonction de la charge du système :

1class AdaptiveTaskProcessor {
2  constructor(options = {}) {
3    this.minConcurrency = options.minConcurrency || 1;
4    this.maxConcurrency = options.maxConcurrency || 20;
5    this.currentConcurrency = options.initialConcurrency || 5;
6    this.running = 0;
7    this.queue = [];
8    this.successCount = 0;
9    this.errorCount = 0;
10    this.adjustInterval = setInterval(() => this.adjustConcurrency(), 5000);
11  }
12
13  adjustConcurrency() {
14    const errorRate = this.errorCount / (this.successCount + this.errorCount || 1);
15
16    if (errorRate > 0.1 && this.currentConcurrency > this.minConcurrency) {
17      this.currentConcurrency = Math.max(this.minConcurrency, this.currentConcurrency - 2);
18      console.log(`Taux d'erreur élevé (${(errorRate * 100).toFixed(0)}%) - réduction de la concurrence à ${this.currentConcurrency}`);
19    } else if (errorRate < 0.02 && this.currentConcurrency < this.maxConcurrency) {
20      this.currentConcurrency = Math.min(this.maxConcurrency, this.currentConcurrency + 1);
21      console.log(`Taux d'erreur faible - augmentation de la concurrence à ${this.currentConcurrency}`);
22    }
23
24    // Réinitialiser les compteurs
25    this.successCount = 0;
26    this.errorCount = 0;
27    this.process();
28  }
29
30  async add(task) {
31    return new Promise((resolve, reject) => {
32      this.queue.push({ task, resolve, reject });
33      this.process();
34    });
35  }
36
37  process() {
38    while (this.running < this.currentConcurrency && this.queue.length > 0) {
39      const { task, resolve, reject } = this.queue.shift();
40      this.running++;
41
42      Promise.resolve()
43        .then(() => task())
44        .then(result => {
45          this.successCount++;
46          resolve(result);
47        })
48        .catch(error => {
49          this.errorCount++;
50          reject(error);
51        })
52        .finally(() => {
53          this.running--;
54          this.process();
55        });
56    }
57  }
58
59  destroy() {
60    clearInterval(this.adjustInterval);
61  }
62}
63
64// Utilisation
65const processor = new AdaptiveTaskProcessor({ minConcurrency: 2, maxConcurrency: 15, initialConcurrency: 5 });
66
67const dinoIds = Array.from({ length: 200 }, (_, i) => i + 1);
68const tasks = dinoIds.map(id => () => processSingleDinosaur(id));
69
70Promise.allSettled(tasks.map(t => processor.add(t))).then(results => {
71  const successful = results.filter(r => r.status === 'fulfilled').length;
72  console.log(`${successful}/${results.length} dinosaures traités avec succès`);
73  processor.destroy();
74});

Résumé

L'exécution parallèle de tâches asynchrones est un outil clé pour optimiser les performances des applications JavaScript. Au Parc Jurassique, où la surveillance et la gestion de multiples systèmes et dinosaures nécessitent des réponses rapides, un bon usage des techniques de traitement parallèle peut considérablement améliorer la réactivité et l'évolutivité du système.

Nous avons appris plusieurs techniques et patterns importants :

1. Promise.all() - l'outil fondamental pour exécuter de nombreuses tâches asynchrones en parallèle
2. Contrôler le nombre d'opérations parallèles - techniques pour limiter la concurrence maximale
3. Traitement par lots - traiter les données par lots pour gérer la mémoire et les limites d'API
4. Files de tâches à priorité - exécuter d'abord les tâches les plus importantes
5. Traitement en flux (streaming) - traiter les données au fur et à mesure pour une UI réactive
6. Worker Threads et Web Workers - exécuter des tâches dans des threads séparés pour les calculs gourmands en CPU
7. Gestion adaptative des ressources - ajuster dynamiquement le niveau de parallélisme selon les taux d'erreur observés

Le choix de la bonne technique dépend du cas d'usage précis, des exigences de performance et des ressources disponibles. Souviens-toi que plus de parallélisme ne signifie pas toujours de meilleures performances - trop d'opérations parallèles peut surcharger le système et, par conséquent, dégrader les performances globales.

En pratique, les meilleurs résultats s'obtiennent souvent en expérimentant différentes stratégies et niveaux de parallélisme, en surveillant les performances et en ajustant les paramètres en conséquence.

1// Hoisting Demonstration - Élévation des déclarations
2console.log("HOISTING EN JAVASCRIPT\n");
3
4console.log("=== HOISTING DES VARIABLES ===\n");
5
6// var est hoisted
7console.log("1. dinosaurName before declaration:", dinosaurName); // undefined (pas une error !)
8var dinosaurName = "T-Rex";
9console.log("2. dinosaurName après declaration:", dinosaurName); // T-Rex
10
11// The above code is interpreted as:
12// var dinosaurName; // declaration hoisted to the top
13// console.log(dinosaurName); // undefined
14// dinosaurName = "T-Rex"; // assignation
15
16console.log("\n=== HOISTING let/const ===\n");
17
18// let i const are in the "temporal dead zone"
19try {
20  console.log(species); // ReferenceError!
21  let species = "Velociraptor";
22} catch (error) {
23  console.log("Error:", error.message);
24}
25
26console.log("\n=== HOISTING DES FONCTIONS ===\n");
27
28// Function declaration is fully hoisted
29console.log("Call before declaration:");
30feedDinosaur("T-Rex"); // Works!
31
32function feedDinosaur(name) {
33  console.log(`Feeding ${name}`);
34}
35
36console.log("\nCall after declaration:");
37feedDinosaur("Triceratops");
38
39console.log("\n=== FUNCTION EXPRESSION ===\n");
40
41// Function expression n'est PAS hoisted
42try {
43  checkSecurity("Secteur A"); // TypeError!
44} catch (error) {
45  console.log("Error:", error.message);
46}
47
48var checkSecurity = function(sector) {
49  console.log(`Vérification de ${sector}`);
50};
51
52// Now works
53checkSecurity("Secteur B");
54
55console.log("\n=== PRACTICAL EXAMPLE ===\n");
56
57function demonstrateHoisting() {
58  console.log("La fonction démarre");
59  console.log("Value of x:", x); // undefined
60
61  if (true) {
62    var x = "Je suis var - function scoped";
63    let y = "Je suis let - block scoped";
64    const z = "Je suis const - block scoped";
65
66    console.log("Dans le bloc - x:", x);
67    console.log("Dans le bloc - y:", y);
68    console.log("Dans le bloc - z:", z);
69  }
70
71  console.log("After block - x:", x); // Works!
72  // console.log("Après le bloc - y:", y); // ERROR! y est block scoped
73  // console.log("Après le bloc - z:", z); // ERROR! z est block scoped
74}
75
76demonstrateHoisting();
77
78console.log("\n=== BEST PRACTICES ===");
79console.log("Use const/let instead of var");
80console.log("Declare variables at the beginning of scope");
81console.log("Use function declarations for global functions");
82console.log("Use arrow functions for local functions");