diff --git a/client/__tests__/util/promiseLimit.test.js b/client/__tests__/util/promiseLimit.test.js new file mode 100644 index 00000000..589c4acf --- /dev/null +++ b/client/__tests__/util/promiseLimit.test.js @@ -0,0 +1,73 @@ +import { PromiseLimit } from "../../src/util/promiseLimit"; +import { range } from "../../src/util/range"; + +const delay = t => new Promise((resolve, reject) => setTimeout(resolve, t)); + +describe("PromiseLimit", () => { + test("simple evaluation, concurrency 1", async () => { + const plimit = new PromiseLimit(1); + const result = await Promise.all([ + plimit.add(() => Promise.resolve(1)), + plimit.add(() => Promise.resolve(2)), + plimit.add(() => Promise.resolve(3)), + plimit.add(() => Promise.resolve(4)) + ]); + expect(result).toEqual([1, 2, 3, 4]); + }); + + test("simple evaluation, concurrency > 1", async () => { + const plimit = new PromiseLimit(100); + const result = await Promise.all([ + plimit.add(() => Promise.resolve(1)), + plimit.add(() => Promise.resolve(2)), + plimit.add(() => Promise.resolve(3)), + plimit.add(() => Promise.resolve(4)) + ]); + expect(result).toEqual([1, 2, 3, 4]); + }); + + test("eval in order of insertion", async () => { + const plimit = new PromiseLimit(100); + let counter = 0; + const result = await Promise.all([ + plimit.add(() => Promise.resolve((counter += 1))), + plimit.add(() => Promise.resolve((counter += 1))), + plimit.add(() => Promise.resolve((counter += 1))), + plimit.add(() => Promise.resolve((counter += 1))) + ]); + expect(result).toEqual([1, 2, 3, 4]); + }); + + test("obeys concurrency limit", async () => { + const plimit = new PromiseLimit(2); + let running = 0; + let maxRunning = 0; + + const cbfn = async i => { + running = running + 1; + maxRunning = running > maxRunning ? running : maxRunning; + await delay(100); + running = running - 1; + }; + + const result = await Promise.all( + range(10).map(i => plimit.add(() => cbfn(i))) + ); + expect(maxRunning).toEqual(2); + }); + + test("rejection", async () => { + const plimit = new PromiseLimit(2); + const result = await Promise.all([ + plimit.add(() => Promise.resolve("OK")), + plimit.add(() => Promise.reject("not OK")).catch(e => e), + plimit.add(() => Promise.resolve("OK")), + plimit + .add(() => { + throw new Error("not OK"); + }) + .catch(e => e.message) + ]); + expect(result).toEqual(["OK", "not OK", "OK", "not OK"]); + }); +}); diff --git a/client/src/actions/index.js b/client/src/actions/index.js index f199e616..8f23057a 100644 --- a/client/src/actions/index.js +++ b/client/src/actions/index.js @@ -7,6 +7,7 @@ import { doBinaryRequest, dispatchNetworkErrorMessageToUser } from "../util/actionHelpers"; +import { PromiseLimit } from "../util/promiseLimit"; import { requestReembed, reembedResetWorldToUniverse } from "./reembed"; /* @@ -15,28 +16,29 @@ we don't need. */ function obsAnnotationFetchAndLoad(dispatch, schema) { const obsAnnotations = schema?.schema?.annotations?.obs ?? {}; - const columns = obsAnnotations.columns ?? []; const index = obsAnnotations.index ?? false; + const columns = (obsAnnotations.columns ?? []).filter( + col => col.name !== index + ); + + const plimit = new PromiseLimit(4); return Promise.all( - columns - .filter(col => col.name !== index) - .map(col => { + columns.map(col => + plimit.add(() => { const path = `annotations/obs?annotation-name=${encodeURIComponent( col.name )}`; const url = `${globals.API.prefix}${globals.API.version}${path}`; - return doBinaryRequest(url); - }) - .map(rqst => rqst.then(buffer => Universe.matrixFBSToDataframe(buffer))) - .map(resp => - resp.then(df => + return doBinaryRequest(url).then(buffer => { + const df = Universe.matrixFBSToDataframe(buffer); dispatch({ type: "universe: column load success", dim: "obsAnnotations", dataframe: df - }) - ) - ) + }); + }); + }) + ) ); } diff --git a/client/src/util/promiseLimit.js b/client/src/util/promiseLimit.js new file mode 100644 index 00000000..113a6348 --- /dev/null +++ b/client/src/util/promiseLimit.js @@ -0,0 +1,50 @@ +/* +This class provides a means to resolve Promises asynchronously, +and limit the number concurrently executed. For example, if +you want to involve a large number of API endpoints using fetch(), +but want to limit the number simultaneously outstanding. + +Example usage: + +const plimit = new PromiseLimit(2); +return Promise.all([ + plimit.add(() => fetch('/foo')), + plimit.add(() => fetch('/bar')), + plimit.add(() => fetch('/baz')) +]) +*/ +export class PromiseLimit { + constructor(maxConcurrency) { + this.queue = new Set(); + this.maxConcurrency = maxConcurrency; + this.pending = 0; + } + + add(fn, ...args) { + // fn - must return a promise + // args - will be passed to fn + return new Promise((resolve, reject) => { + this.queue.add({ fn, args, resolve, reject }); + this._resolveNext(false); + }); + } + + _resolveNext = (completed = true) => { + if (completed) this.pending -= 1; + + while (this.queue.size > 0 && this.pending < this.maxConcurrency) { + const task = this.queue.values().next().value; // order of insertion + this.pending += 1; + this.queue.delete(task); + const { resolve, reject, fn, args } = task; + + try { + const result = fn(...args); + result.then(this._resolveNext, this._resolveNext); + result.then(resolve, reject); + } catch (err) { + reject(err); + } + } + }; +}