runWithLimitedConcurrency.js

runWithLimitedConcurrency.js

  /**
   * https://timtech.blog/posts/limiting-async-operations-promise-concurrency-javascript/
   * @param tasks an array of async functions taking zero parameters and returning an array of the same (representing downstream work to be added to the queue)
   * @param CONCURRENT_WORKERS
   * @return {Promise<void>}
   */
  async function runWithLimitedConcurrency (tasks, CONCURRENT_WORKERS = navigator.hardwareConcurrency) {
    const allDone = Promise.withResolvers()
    const waitingWorkers = new Set()

    function resumeWaitingWorkers () {
      // avoid messing the iterator with add/delete in the loop
      const waitListCopy = [...waitingWorkers]
      for (const worker of waitListCopy) {
        // let the async slip to get concurrency
        worker()
      }
    }

    function createWorker (next_, _id) {
      const worker = async () => {
        waitingWorkers.delete(worker)
        try {
          let task
          while ((task = next_())) {
            const result = await task()
            if (Array.isArray(result) && result.length) {
              tasks.push(...result)
              resumeWaitingWorkers()
            }
          }
        } catch (err) {
          console.log(err)
          console.log('logging and resuming')
        }
        waitingWorkers.add(worker)
        if (waitingWorkers.size === CONCURRENT_WORKERS) {
          allDone.resolve(undefined)
        }
      }
      return worker
    }

    for (let i = 0; i < CONCURRENT_WORKERS; i++) {
      waitingWorkers.add(createWorker(tasks.pop.bind(tasks), i))
    }
    resumeWaitingWorkers()
    await allDone.promise
    console.log('work done')
  }

  await runWithLimitedConcurrency(pdfs.map(file => (async () => convertPDF(file, temp_dir))))