Add fetch concurrency limit on obs annotation loading (#1318)

* add fetch concurrency limit

* add tests for PromiseLimit

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"]);
+	});
+});

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
-          })
-        )
-      )
+          });
+        });
+      })
+    )
   );
 }
 

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);
+			}
+		}
+	};
+}