Build a request queue with promises?

Question

My goal is to run a data import against an REST endpoint.

What I don't want is to wait for a request to be resolved before I fire a new one. I want to 'simulate' parallel connections.

And I am not sure I have a basic knowledge problem here.

This code creates child processes:

const numchild = require('os').cpus().length;
const SIZE = 1000;
const SIZE_PER_CHILD = SIZE / numchild;

for (let i = 0; i < numchild; i++) {
  const child = child_process.fork('./child.js');
  child.send({ start: i * SIZE_PER_CHILD, end: (i + 1) * SIZE_PER_CHILD });
  // Some more code...
}

Then per child process I want to generate a random payload for importing and fire it against the REST endpoint:

process.on('message', async function({ start, end }) {
  const count = start;
  while (count < end) {
    const generatedData = 'I was generated! Yay!' + count;
    await axios.put('/api/v1/import', generatedData);
    count++;
  }
});

The above approach would wait for every import request to finish and then fire the next one, until all child imports are done. Not what I want.

Now the endpoint I am firing against should be capable of handling more than the requests I am able to generate like this.

I can rewrite it like this:

process.on('message', async function({ start, end }) {
  const count = start;
  while (count < end) {
    const generatedData = 'I was generated! Yay!' + count;
    axios.put('/api/v1/import', generatedData).then(() => console.log('I am done with this one'));
    count++;
  }
});

The problem with this approach of course would be that all requests are generated within a few seconds and are fired against the endpoint. This is more like DOS-style, I guess. Also not what I want.

What I am hoping to achieve is something like: Have 15 open connections per child process. If a request finished, queue the next request until you have 15 requests pending again.

So I tried this:

process.on('message', async function({ start, end }) {
  const count = start;
  let queue = [];
  while (count < end) {
    if (queue.length === 15) {
      queue = queue.filter(async (promise) => {
        const state = await promiseState(promise);
        return state !== 'fulfilled';
      });
    } else {
      const generatedData = 'I was generated! Yay!' + count;
      queue.push(axios.put('/api/v1/import', generatedData).then(() => console.log('I am done with this one')));
      count++;
    }
  }
});

function promiseState(p) {
  const t = {};
  return Promise.race([p, t])
    .then(v => (v === t) ? "pending" : "fulfilled", () => "rejected");
}

Also doesn't work and doesn't make sense, right? The filter function returns promises and therefore what I am trying to do doesn't work.

Any way I can achieve this?

Answer 1

Try a p-queue. Below the concurrency is set to 3, meaning max 3 calls are executed at once within this queue:

import PQueue from 'p-queue';

const queue = new PQueue({
  concurrency: 3,
});

process.on('message', async function ({ start, end }) {
  var calls = [];
  var count = start;
  while (count < end) {
    const generatedData = 'I was generated! Yay!' + count;

    calls.push(
      queue.add(() => {
        return axios
          .put('/api/v1/import', generatedData)
          .then(() => console.log('I am done with this one'));
      })
    );
    count++;
  }
  var results = await Promise.all(calls);
});