fix: throttle the rate AND number of request_state msg'es from the frontend

[maartenbreddels]

Fixes #534

Related: jupyter-widgets/ipywidgets#2765

@jasongrout this is a variant of your mapBatch that will try to put 'room' number of messages in flight, so once 1 message comes in (and the rate is not too high), the next message will be directly sent. This should decrease latency.

Let me know what you think, we probably want this functionality to go into jupyter-widgets.

[jasongrout]

This package is still using bluebird promises and hasn't been updated in 3 years.

[maartenbreddels]

Yes, I'm not happy with that package either, looking for something else. https://github.com/sindresorhus/p-queue may actually do all of this..

[maartenbreddels]

https://www.npmjs.com/package/p-limit is a bit easier to understand than p-queue.

[maartenbreddels]

I thought about it a bit more, my argument was that my method would be faster with latency. But with 2000 widgets, at 100 per batch, you get an extra 20x the latency, which could be 2 seconds? On a total time of maybe 20 seconds?

So I think in terms of performance, we don't gain much actually.

[maartenbreddels]

Ok, apart from the performance, we may still hit the iopub rate limit without this p-limit.

[jasongrout]

What about this implementation of concurrent batching?

/**
 * A batched map implementation with concurrency
 *
 * Take a list, a batch size, a mapping function, and a max concurrency
 */
export async function mapBatch(
  list: any[],
  batchSize: number,
  concurrency: number,
  fn: Function,
  thisArg: any
): Promise<any[]> {
  const results = new Array(list.length);

  const chunks: PromiseClosure[] = [];

  // Process a single chunk and resolve to the next chunk if available
  async function processChunk(chunk: any[], start: number): Promise<void> {
    const chunkResult = await Promise.all(
      chunk.map((v, i) => fn.call(thisArg, v, start + i, list))
    );

    // Splice the chunk results into the results array. We use results.length
    // because the last chunk may not be a full batchSize.
    results.splice(start, chunkResult.length, ...chunkResult);

    // Start the next work item by returning its promise
    if (chunks.length > 0) {
      return chunks.shift()!();
    }
  }

  // Make closures for each batch of work
  for (let i = 0; i < list.length; i += batchSize) {
    chunks.push(() => processChunk(list.slice(i, i + batchSize), i));
  }

  // Start the concurrent chunks. Each chunk will automatically start
  // the next available chunk when it finishes.
  await Promise.all(chunks.splice(0, concurrency).map(f => f()));
  return results;
}

[jasongrout]

What about this implementation of concurrent batching?

See the cleaned-up implementation at jupyter-widgets/ipywidgets#2765

[jasongrout]

Actually, I don't think we need throttling. Having a concurrency of, say, 100 will naturally introduce the appropriate amount of throttling, i.e., the network roundtrip. Since all we care about is not hitting the zmq socket with more than 100 requests or so simultaneously, concurrency naturally throttles to about that.

[jasongrout]

Actually, I don't think we need throttling.

Ah, the throttling is for the separate iopub rate limiting. Yes, good point.

[SylvainCorlay]

Merging this!