Limited to parallelism with async / wait in Typescript / ES7

Question

I experimented a bit with Typescript, but now I'm a little fixated on how to use async / await efficiently.

I am inserting a bunch of records into the database and I need to get a list of identifiers returned by each insert. The following simplified example works in general, but it is not as elegant as we would like, and it is completely consistent.

async function generatePersons() {
    const names = generateNames(firstNames, lastNames);
    let ids = []
    for (let name of names) {
        const id = await db("persons").insert({
            first_name: name.firstName,
            last_name: name.lastName,
        }).returning('id');
        ids.push(id[0])
    }
    return ids
}

I tried using mapto avoid creating a list idsmanually, but I could get this to work.

What I would like to have is a limited amount of parallelism. Therefore, my asynchronous calls must be executed in parallel to a certain limit, for example. I would just like to have 10 open queries, but no more.

parallelism async/await Typescript Javascript ES7? , ?

_{PS: , , , . , , ,}

Answer 1

Promise.all , .

, , . , , . - ( !)

async function asyncThrottledMap<T, U>(maxCount: number, array: T[], f: (x: T) => Promise<U>) {
    let inFlight = new Set<Promise<U>>();
    const result: Promise<U>[] = [];

    // Sequentially add a Promise for each operation.
    for (let elem of array) {

        // Wait for any one of the promises to complete if there are too many running.
        if (inFlight.size >= maxCount) {
            await Promise.race(inFlight);
        }

        // This is the Promise that the user originally passed us back.
        const origPromise = f(elem);
        // This is a Promise that adds/removes from the set of in-flight promises.
        const handledPromise = wrap(origPromise);
        result.push(handledPromise);
    }

    return Promise.all(result);

    async function wrap(p: Promise<U>) {
        inFlight.add(p);
        const result = await p;
        inFlight.delete(p);
        return result;
    }
}

, inFlight - , .

result - Promise s. promises inFlight. , Promise.race.

, .

Answer 2

async-parallel, , . , :

async function generatePersons(): Promise<number[]> {
    const names = generateNames(firstNames, lastNames);
    return await Parallel.map(names, async (name) => 
        await db("persons").insert({
            first_name: name.firstName,
            last_name: name.lastName,
        }).returning('id'));
}

, , ...

Parallel.concurrency = 4;

Answer 3

, . , :

  // Goes over the "data" array, calls and waits for each "task" and processes "runnerCount" tasks in parallel
  function inParallel(task, data, runnerCount) {
    let i = 0, results = [];

   async function runner() {
      while(i < data.length) {
         const pos = i++; // be aware: concurrent modification of i
         const entry = data[pos]; 
         results[pos] = await task(entry);
      }
   }

    const runners = Array.from({ length: runnerCount }, runner);

    return Promise.all(runners).then(() => results);
 }

:

  const delay = ms => new Promise(res => setTimeout(res, ms));

 inParallel(async time => {
   console.log('Timer for ${time}ms starts');
   await delay(time);
   console.log('Timer for ${time}ms ends');
 }, [5000, 6000, 1000]/*ms*/, 2/*in parallel*/);

Answer 4

, Jonas , . :

1. ( ).
2. , , , , Promise.all .

function parallelMap(values, workFn, maxConcurrency = 2) {
  const length = values.length;
  const results = Array.from({ length });

  let pos = 0;
  let completed = 0;

  return new Promise(resolve => {
    function work() {
      if (completed === length) {
        return resolve(results);
      }

      if (pos >= length) {
        return;
      }

      const workIndex = pos;
      const item = values[workIndex];
      pos = pos + 1;

      return workFn(item, workIndex)
        .then(result => {
          results[workIndex] = result;
          completed = completed + 1;
          work();
        })
        .catch(result => {
          results[workIndex] = result;
          completed = completed + 1;
          work();
        });
    }

    for (let i = 0; i < maxConcurrency; i++) {
      work();
    }
  });
}

:

async function fakeRequest({ value, time = 100, shouldFail = false }) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      if (shouldFail) {
        reject("Failure: " + value);
      } else {
        resolve("Success: " + value);
      }
    }, time);
  });
}

test("basic 'working' prototype", async () => {
  const values = [1, 2, 3, 4, 5, 6];
  const results = await parallelMap(values, value => {
    return fakeRequest({ value, time: 100, shouldFail: value % 2 === 0 });
  });

  expect(results).toEqual([
    "Success: 1",
    "Failure: 2",
    "Success: 3",
    "Failure: 4",
    "Success: 5",
    "Failure: 6"
  ]);
}, 350); // each task takes ~100ms to complete, 6 tasks, two at a time = ~300 ms

.

Answer 5

Is there a reasonably elegant way to achieve such limited parallelism using async / await in Typescript or Javascript ES7

You will need to use Promise.all. those. collect all the promises in the array and await Promise.all([all,the,stuff]).

More details

https://developer.mozilla.org/en/docs/Web/JavaScript/Reference/Global_Objects/Promise/all