Parallel class for javascript promises

Question

I created a class to run a certain number of tasks in parallel in the browser. The constructor takes an array of inputs, a function that takes a single input and returns a promise, and a count of the number of tasks to run in parallel. This is for doing some web scraping so I don't want to create a hundred ajax requests and wait for them to all complete with Promise.all, besides worrying about a timeout or failure I don't want to thrash the connections to a server (jsfiddle):

// run 5 at a time, calling func on each element of delays array
let par = new Parallel(delays, func, 5);
par.run().then(results => { ... });

It runs all the items and resolves if there were no errors, or rejects if there were any errors. The results array contains each result or error from a reject. Is there a better way?

class Parallel {

  constructor(inputs, func, count) {
    this.inputs = inputs;
    this.func = func;
    this.count = count;
    this.index = 0;
    this.active = 0;
    this.enabled = true;
    this.hadError = false;
  }

  cancel() {
    this.enabled = false;
  }

  run() {
    this.start = performance.now();

    return new Promise( (resolve, reject) => {
      this.results = [];

      let checkDone = () => {
        if (this.active === 0 && (this.index >= this.inputs.length || (!this.enabled))) {
          this.end = performance.now();
          if (this.hadError)
            reject(this.results);
          else
            resolve(this.results);
          return true;
        }
      }

      let setAndCheck = (index, result) => {
        this.active--;
        this.results[index] = result;
        queue();
      }

      let handleOne = (index, input) => {
        this.active++;
        this.func(input)
        .then(result => setAndCheck(index, result))
        .catch(err => {
          this.hadError = true;
          console.log('Error with input:', input);
          console.error(err);
          setAndCheck(index, err);
        });
      }

      let queue = () => {
        if (checkDone()) return;
        while (this.enabled && (this.active < this.count) && this.index < this.inputs.length) {
          let index = this.index;
          let input = this.inputs[index];
          this.index = this.index + 1;
          handleOne(index, input);
        }
      }

      queue();
    });
  }
}

let totalDelay = 0;
let func = (delay) => new Promise(resolve => setTimeout(() => resolve(delay), delay));
let delays = [];
for (var i = 0; i < 20; i++) {
	let d = Math.random() * 1000 + 500;
	delays.push(d);
  totalDelay += d;
}

let start = performance.now();
let par = new Parallel(delays, func, 5);
par.run().then(results => {
	var total = results.reduce((acc, val) => acc + val, 0);
  var actual = performance.now() - start;
  document.body.innerHTML = `<ul>
  <li>totalDelay: ${totalDelay.toFixed(2)}</li>
  <li>actual: ${actual.toFixed(2)}</li>
</ul>`;
});

Answer 1

I created a class to run a certain number of tasks in parallel in the browser.

Unless you're using web workers, in JS, nothing actually runs in parallel. The external processes (file reads, network requests, etc) may be running in parallel after spawning them, but JS still launches and completes in some sequence, asynchronously.

It runs all the items and resolves if there were no errors, or rejects if there were any errors.

Sounds like you're looking for Promise.all. It takes an array of promises, and resolves if all of them resolve or rejects if any one of them rejects.

Now I've seen code that's eerily similar to this one, and I tell you it's a pain to debug. You have a bunch of functions depending on counters "flag properties" that alter how the functions work. Since async doesn't guarantee order, the values aren't guaranteed, therefore how the function runs is also not guaranteed. This then makes your queue hard to test and you end up doing "hope-driven development" - a project devoid of unit tests.

Hopefully I've scared you, since I don't ever want to see queueing done this way. Makes you lose hair. What I suggest you do is just take in an array of inputs, recursively work on them by chunks, concatenate the results and hand it over. No counters, no flags.

Here's a simple, no-cancel version of the queue. It launches the queue chunkSize at a time. Then it launches the rest of the array, up until the array is exhausted. Then the chunks are concatenated. All you need for inputs are functions that return promises. How you construct them or feed them input is up to you, not the queue. Below shows how to use closures. Alternatively, you can use bind.

const launchQueue = (q, chunkSize) => {
  // No more items, resolve with an empty array
  if (!q.length) return Promise.resolve([]);

  // Slice the array
  const chunk = q.slice(0, chunkSize);
  const remain = q.slice(chunkSize);

  // Call each function and store the promises returned to an array.
  // Promise.all fails fast. Catch errors and resolve with those.
  const chunkPromises = chunk.map(f => f.call().then(r => r, e => e));

  // Wait for all to resolve  
  return Promise.all(chunkPromises).then(r => {

    // Run the remaining items
    return launchQueue(remain, chunkSize).then(s => {

      // Concat results.
      return r.concat(s);
    });
  });
}

// For each input, create a function that returns a promise that
// that resolves when the operation completes.
const queue = [9000, 7000, 4000, 5000, 8000, 1000, 3000, 2000, 4000, 6000].map(n => {
  return () => new Promise(r => {
    console.log(`fn with ${n} requested`);
    setTimeout(() => {
      console.log(`fn with ${n} resolved`);
      
      // Prove that whatever we resolve here ends up in the
      // call below.
      r(-n);
    }, n);
  });
});

// Hand over the functions to the queueing function and wait for
// the results.
launchQueue(queue, 2).then(r => console.log(r));

An async-await version is much more readable:

const launchQueue = async (q, chunkSize) => {
  if (!q.length) return [];

  const chunk = q.slice(0, chunkSize);
  const remain = q.slice(chunkSize);
  // The then is much more concise than a try-catch.
  const chunkPromises = chunk.map(f => f.call().then(r => r, e => e));
  const chunkResults = await Promise.all(chunkPromises);
  const remainResults = await launchQueue(remain, chunkSize);
  return chunkResults.concat(remainResults);
}

// For each input, create a function that returns a promise that
// that resolves when the operation completes.
const queue = [9000, 7000, 4000, 5000, 8000, 1000, 3000, 2000, 4000, 6000].map(n => {
  return () => new Promise(r => {
    console.log(`fn with ${n} requested`);
    setTimeout(() => {
      console.log(`fn with ${n} resolved`);
      
      // Prove that whatever we resolve here ends up in the
      // call below.
      r(-n);
    }, n);
  });
});

// Hand over the functions to the queueing function and wait for
// the results.
launchQueue(queue, 2).then(r => console.log(r));