Parallel Execution of Asynchronous Tasks

In Jurassic Park, simultaneously managing many systems and monitoring dozens of dinosaurs requires an efficient approach to running many tasks at the same time. In this exercise we will focus on techniques for parallel execution of asynchronous tasks in JavaScript, which will allow us to increase the performance of our park management systems.

Why Is Parallel Execution Important?

Imagine we need to check the status of all 50 dinosaur enclosures before opening the park. If each check takes about 1 second and we execute them sequentially (one after another), the entire process will take at least 50 seconds — far too long before letting visitors in!

By executing these tasks in parallel, we can significantly reduce the total time needed to perform all operations, limiting it to the duration of the single longest operation (plus some overhead).

Promise.all() — The Basic Tool for Parallel Execution

As we saw in previous lessons,

Promise.all()

is the fundamental tool for running many asynchronous tasks in parallel:

1async function checkAllEnclosureStatuses() {
2  const enclosureList = await fetchEnclosureList();
3  console.log(`Checking status of ${enclosureList.length} enclosures...`);
4
5  const startTime = Date.now();
6
7  // Launch checking of all enclosures in parallel
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(`Checking all enclosures took ${executionTime.toFixed(2)} seconds`);
14
15  // Analyze results
16  const secureEnclosures = results.filter(w => w.status === 'secure');
17  const problematicEnclosures = results.filter(w => w.status !== 'secure');
18
19  console.log(`Bezpieczne enclosurei: ${secureEnclosures.length}`);
20  console.log(`Problematyczne enclosurei: ${problematicEnclosures.length}`);
21
22  if (problematicEnclosures.length > 0) {
23    console.warn("Problems found in the following enclosures:");
24    problematicEnclosures.forEach(w => {
25      console.warn(`- Enclosure #${w.id}: ${w.status} (${w.message})`);
26    });
27  }
28
29  return {
30    total: results.length,
31    secure: secureEnclosures.length,
32    problematic: problematicEnclosures,
33    czasWykonania
34  };
35}

In the above example, we use

Promise.all()

to check the status of all enclosures in parallel. Instead of waiting for each check to complete before starting the next, all checks are initiated simultaneously and we wait for all of them to complete.

Challenges of Parallel Execution

Parallel execution of asynchronous tasks brings many benefits, but also comes with certain challenges:

1. Resource Management

Launching too many operations in parallel can overload available system resources (memory, network connections, etc.).

2. Error Handling

Remember that

Promise.all()

will fail if any one promise is rejected. With a large number of parallel operations, a single error can cause the entire process to fail.

3. Too Much Parallelism

Running all tasks in parallel is not always the best solution. Sometimes we need to control the maximum number of parallel operations.

Controlling Concurrency — Not Too Many at Once

Running too many operations simultaneously can overload the server or API. We need a way to run tasks in parallel but with a limit on how many run at the same time:

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// Check all paddocks but maximum 5 at a time
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}

Libraries for Managing Parallelism

Several popular libraries make managing parallel task execution easier:

p-limit

The p-limit library allows easy limiting of parallel operations:

1const pLimit = require('p-limit');
2
3async function monitorWszystkieDinosaury() {
4  const dinosaurList = await fetchDinosaurList();
5
6  // Create a limiter instance with at most 5 parallel operations
7  const limit = pLimit(5);
8
9  // Create a promise array with limited parallelism
10  const promises = dinosaurList.map(dino => {
11    return limit(() => monitorDinosaura(dino.id));
12  });
13
14  // Wait for all operations to complete
15  const results = await Promise.all(promises);
16
17  return results;
18}

Parallel Processing with Error Handling

As we mentioned earlier,

Promise.all()

will fail if any one promise is rejected. In some cases we want to continue processing even if some operations fail.

In ECMAScript 2020,

Promise.allSettled()

was introduced, which solves this problem:

1async function runDiagnosticsForAllSystems() {
2  const systems = await fetchAllSystems();
3
4  console.log(`Running diagnostics for ${systems.length} systems...`);
5
6  // Launch diagnostics for all systems in parallel and wait for all to complete
7  const results = await Promise.allSettled(
8    systems.map(async (system) => {
9      try {
10        console.log(`Diagnostics for system ${system.name}...`);
11        const result = await systemDiagnostics(system.id);
12        console.log(`Diagnostics for system ${system.name} completed`);
13        return { systemId: system.id, status: 'success', result: result };
14      } catch (error) {
15        console.error(`Diagnostics error for system ${system.name}:`, error);
16        return { systemId: system.id, status: 'error', error: error.message };
17      }
18    })
19  );
20
21  // Analyze results
22  const udata = results.filter(w => w.status === 'fulfilled' && w.value.status === 'success');
23  const nieudata = results.filter(w => w.status === 'rejected' || (w.status === 'fulfilled' && w.value.status === 'error'));
24
25  console.log(`Diagnostics completed: ${udata.length} successful, ${nieudata.length} niesuccessful`);
26
27  return {
28    total: results.length,
29    udata: udata.length,
30    nieudata: nieudata.length,
31    results
32  };
33}

Batch Processing

Sometimes it is better to process large collections of items in batches:

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(`Processing batch ${Math.floor(i / batchSize) + 1}: dinosaurs ${i + 1}-${Math.min(i + batchSize, dinosaurIds.length)}`);
7
8    // Process the batch in parallel
9    const batchResults = await Promise.all(
10      batch.map(id => processSingleDinosaur(id))
11    );
12
13    results.push(...batchResults);
14
15    // Optional: small pause between batches to avoid API rate limits
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// Usage
30const allDinoIds = Array.from({ length: 100 }, (_, i) => i + 1);
31processDinosaursBatch(allDinoIds, 10).then(results => {
32  console.log(`Processed ${results.length} dinosaurs`);
33  const avgHealth = results.reduce((sum, d) => sum + d.health, 0) / results.length;
34  console.log(`Average health: ${avgHealth.toFixed(1)}%`);
35});

Priority Queue

Some operations are more urgent than others. A priority queue ensures critical tasks run first:

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// Usage
41const parkTaskQueue = new PriorityTaskQueue(4);
42
43// High priority — security alerts
44async function handleSecurityAlert(alertId) {
45  return parkTaskQueue.add(
46    () => processSecurityAlert(alertId),
47    'high'
48  );
49}
50
51// Medium priority — regular monitoring
52async function scheduledMonitoring(paddockId) {
53  return parkTaskQueue.add(
54    () => checkSinglePaddock(paddockId),
55    'medium'
56  );
57}
58
59// Low priority — data sync
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 Results

When processing many items, you can stream results as they complete rather than waiting for all:

1async function* processWithStreaming(items, processor) {
2  const promises = items.map((item, index) =>
3    processor(item).then(result => ({ index, result }))
4  );
5
6  // Use a tracker to yield results in completion order
7  const pending = new Map(promises.map((p, i) => [i, p]));
8
9  while (pending.size > 0) {
10    // Wait for the next promise to settle
11    const settled = await Promise.race(pending.values());
12    pending.delete(settled.index);
13    yield settled.result;
14  }
15}
16
17// Usage — see results as they arrive
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}] Dinosaur ${result.id}: ${result.health}% health`);
28  }
29}
30
31monitorDinosaursStreaming([1, 2, 3, 4, 5, 6, 7, 8]);

Worker Threads in Node.js

In Node.js, we can use the

worker_threads

module to perform intensive computations in separate threads:

1const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
2
3// Main code
4if (isMainThread) {
5  async function analyzeDinosaurData() {
6    const data = await fetchLargeDataSets();
7
8    // Split data into parts, one for each CPU core
9    const threadCount = require('os').cpus().length;
10    const dataParts = splitData(data, threadCount);
11
12    // Create a worker for each data part
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(`Worker stopped with exit code ${code}`));
24          }
25        });
26      });
27    });
28
29    // Wait for all workers to complete
30    const results = await Promise.all(workers);
31
32    // Merge results
33    return mergeResults(results);
34  }
35} else {
36  // Code executed in each worker thread
37  const { data, index } = workerData;
38
39  // Perform complex computations on the data
40  const results = processData(data);
41
42  // Send results back to the main thread
43  parentPort.postMessage(results);
44}

Web Workers for CPU-Intensive Work

For heavy computations that would block the main thread in the browser, use Web Workers:

1// main.js — creating a 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 — runs in separate thread
23self.onmessage = function(event) {
24  if (event.data.type === 'ANALYZE') {
25    const { data } = event.data;
26
27    // CPU-intensive computation — does not block the main thread
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};

Adaptive Task Processor

A complete solution that adapts concurrency based on system load:

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(`High error rate (${(errorRate * 100).toFixed(0)}%) — reducing concurrency to ${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(`Low error rate — increasing concurrency to ${this.currentConcurrency}`);
22    }
23
24    // Reset counters
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// Usage
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(`Processed ${successful}/${results.length} dinosaurs successfully`);
73  processor.destroy();
74});

Summary

Parallel execution of asynchronous tasks is a key tool for optimizing JavaScript application performance. In Jurassic Park, where monitoring and managing multiple systems and dinosaurs requires fast responses, proper use of parallel processing techniques can significantly improve the responsiveness and scalability of the system.

We learned several important techniques and patterns:

1. Promise.all() — the fundamental tool for running many asynchronous tasks in parallel
2. Controlling the number of parallel operations — techniques for limiting maximum concurrency
3. Batch processing — processing data in batches to manage memory and API limits
4. Priority task queues — executing more important tasks first
5. Stream processing — processing data as it arrives for responsive UI
6. Worker Threads and Web Workers — executing tasks in separate threads for CPU-intensive work
7. Adaptive resource management — dynamically adjusting the level of parallelism based on observed error rates

The choice of the right technique depends on the specific use case, performance requirements, and available resources. Remember that more parallelism does not always mean better performance — too many parallel operations can lead to system overload and, consequently, to degraded overall performance.

In practice, the best results are often achieved by experimenting with different strategies and parallelism levels, monitoring performance, and adjusting parameters accordingly.