限制在给定时间开放的承诺量

/**
 * Performs a list of callable actions (promise factories) so
 * that only a limited number of promises are pending at any
 * given time.
 *
 * @param listOfCallableActions An array of callable functions,
 *     which should return promises.
 * @param limit The maximum number of promises to have pending
 *     at once.
 * @returns A Promise that resolves to the full list of values
 *     when everything is done.
 */
function throttleActions(listOfCallableActions, limit) {
  // We'll need to store which is the next promise in the list.
  let i = 0;
  let resultArray = new Array(listOfCallableActions.length);

  // Now define what happens when any of the actions completes.
  // Javascript is (mostly) single-threaded, so only one
  // completion handler will call at a given time. Because we
  // return doNextAction, the Promise chain continues as long as
  // there's an action left in the list.
  function doNextAction() {
    if (i < listOfCallableActions.length) {
      // Save the current value of i, so we can put the result
      // in the right place
      let actionIndex = i++;
      let nextAction = listOfCallableActions[actionIndex];
      return Promise.resolve(nextAction()).then(result => {
        // Save results to the correct array index.
        resultArray[actionIndex] = result;
      }).then(doNextAction);
    }
  }

  // Now start up the original <limit> number of promises.
  // i advances in calls to doNextAction.
  let listOfPromises = [];
  while (i < limit && i < listOfCallableActions.length) {
    listOfPromises.push(doNextAction());
  }
  return Promise.all(listOfPromises).then(() => resultArray);
}

// Test harness:

function delay(name, ms) {
  return new Promise((resolve, reject) => setTimeout(() => {
    console.log(name);
    resolve(name);
  }, ms));
}

var ps = [];
for (let i = 0; i < 10; i++) {
  ps.push(() => delay("promise " + i, Math.random() * 3000));
}

throttleActions(ps, 3).then(result => console.log(result));

// throttlep :: Number -> [(* -> Promise)]
const throttlep = n=> Ps=>
  new Promise ((pass, fail)=> {
    // r is the number of promises, xs is final resolved value
    let r = Ps.length, xs = []
    // decrement r, save the resolved value in position i, run the next promise
    let next = i=> x=> (r--, xs[i] = x, run(Ps[n], n++))
    // if r is 0, we can resolve the final value xs, otherwise chain next
    let run = (P,i)=> r === 0 ? pass(xs) : P().then(next(i), fail)
    // initialize by running the first n promises
    Ps.slice(0,n).forEach(run)
  })

// -----------------------------------------------------
// make sure it works

// delay :: (String, Number) -> (* -> Promise)
const delay = (id, ms)=>
  new Promise (pass=> {
    console.log (`running: ${id}`)
    setTimeout(pass, ms, id)
  })

// ps :: [(* -> Promise)]
let ps = new Array(10)
for (let i = 0; i < 10; i++) {
  ps[i] = () => delay(i, Math.random() * 3000)
}

// run a limit of 3 promises in parallel
// the first error will reject the entire pool
throttlep (3) (ps) .then (
  xs => console.log ('result:', xs),
  err=> console.log ('error:', err.message)
)

async function throttle(tasks, max) {
    async function run(_, i) {
        values[i] = await tasks[i]();
        if (max < tasks.length) return run(_, max++);
    };
    const values = [];
    try {
        await Promise.all(tasks.slice(0, max).map(run));
    } catch (error) {
        max = tasks.length; // don't allow new tasks to start
        throw error;
    }
    return values;
}

// Demo
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));

const tasks = Array.from({length: 10}, (_, i) => 
    async () => {
        console.log(`task ${i} starts`);
        await delay((1 + i % 3)*1000);
        console.log(`task ${i} ends with ${i*10}`);
        return i*10;
    }
);

throttle(tasks, 4).then(console.log);

function ParallelMap(array, callbackFn, maxDegreeOfParallelism = -1) {
  if (maxDegreeOfParallelism < -1 || maxDegreeOfParallelism == 0) return Promise.reject(`'maxDegreeOfParallelism' must be either -1 or greater than 0`);

  return new Promise((resolve, reject) => {
    const inputArraySize = array.length;

    let indexTracker = 0;
    let completedTracker = 0;

    const output = new Array(inputArraySize);
    const errors = new Array();

    const processNext = () => {
      const elementIndex = indexTracker++;
      const element = array[elementIndex];

      callbackFn(element, elementIndex, array).then(
        value => output[elementIndex] = value,
        reason => errors.push({
          index: elementIndex,
          error: reason
        })
      ).finally(() => {
        ++completedTracker;

        if (completedTracker == inputArraySize) {
          if (errors.length > 0) reject(errors);

          else resolve(output);
        } else if (indexTracker < inputArraySize) processNext();
      });
    };

    for (let index = 0, count = maxDegreeOfParallelism < 0 ? inputArraySize : Math.min(maxDegreeOfParallelism, inputArraySize); index < count; ++index) {
      processNext();
    }
  });
}



// Usage

(async() => {
  const input = new Array(10).fill(1); // Array containing 10 '1' values

  const oneSecondTask = (value, index) => {
    return new Promise(resolve => {
      setTimeout(() => {
        resolve(value + index); // Extremely complex calculation of adding index to value 1
      }, 1000);
    });
  };

  console.log(`const input = [${input.join(', ')}];`);
  console.log(`---------------------------------------------`);
  console.log(`... wait for 10s ...`);
  console.log(`---------------------------------------------`);

  let start = Date.now();
  let maxDegreeOfParallelism = 1;
  let result = await ParallelMap(input, oneSecondTask, maxDegreeOfParallelism);
  console.log(`const result = [${result.join(', ')}];`);
  console.log(`${(Date.now() - start) / 1000}s to process ${input.length} items (taking 1s each) one at a time`);

  console.log(`---------------------------------------------`);

  start = Date.now();
  maxDegreeOfParallelism = 2;
  result = await ParallelMap(input, oneSecondTask, maxDegreeOfParallelism);
  console.log(`const result = [${result.join(', ')}];`);
  console.log(`${(Date.now() - start) / 1000}s to process ${input.length} items (taking 1s each) in parallel using ${maxDegreeOfParallelism} concurrent tasks`);

  console.log(`---------------------------------------------`);

  start = Date.now();
  maxDegreeOfParallelism = 5;
  result = await ParallelMap(input, oneSecondTask, maxDegreeOfParallelism);
  console.log(`const result = [${result.join(', ')}];`);
  console.log(`${(Date.now() - start) / 1000}s to process ${input.length} items (taking 1s each) in parallel using ${maxDegreeOfParallelism} concurrent tasks`);

  console.log(`---------------------------------------------`);

  start = Date.now();
  maxDegreeOfParallelism = 10;
  result = await ParallelMap(input, oneSecondTask, maxDegreeOfParallelism);
  console.log(`const result = [${result.join(', ')}];`);
  console.log(`${(Date.now() - start) / 1000}s to process ${input.length} items (taking 1s each) in parallel using ${maxDegreeOfParallelism} concurrent tasks`);
})();