Concurrent Streaming Queue

by Matt

JavaScript

/**
 * GOAL:
 * Start jobs as soon as possible, but limit concurrency s.t. no more
 * than N jobs are running in parallel. Completed jobs must be processed
 * in serial order.
 */

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


function createRequest(id) {
  return Promise.resolve()
    //.then(() => console.log("start", id))
    .then(() => sleep(Math.random() * 500 + 500)) // mock network request
    .then(() => console.log("finish", id, performance.now()))
    .then(() => `Result(${id})`); // mock result of request
}

/**
 * https://jakearchibald.com/2017/async-iterators-and-generators/#making-streams-iterate
 *
 * @template T
 * @param {ReadableStream<T>} stream
 */
async function* streamAsyncIterator(stream) {
	const reader = stream.getReader();
	try {
		while (true) {
			const { done, value } = await reader.read();
			if (done) return;
			yield value;
		}
	} finally {
		reader.releaseLock();
	}
}

const stream = new TransformStream()
const reader = stream.readable.getReader()
const writer = stream.writable.getWriter()

async function main() {
  const arr = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  
  const limit = 2
  const promises = []
  
  const canAdd = () => promises.length < limit
  
  const shift = async () => {
    const promise = promises.shift()
    const res = await promise
    console.log("serial", res, performance.now())
    return res
  }
  
  const pending = async () => {
  	if (promises.length < limit) {
			return
    } else {
			return await shift()
    }
  }
  
  
  const readableStream = new ReadableStream({
  	async start(controller) {
      for (const [i, item] of arr.entries()) { // imagine this is a stream of unknown length
        // wait for any pending job if needed
        
        const res = await pending()
        
        if (res) {
        	controller.enqueue(res)
        }
        
        const promise = createRequest(item)
        promises.push(promise)
      }
      
      // drain any remaining promises
     ...