Merge pull request #1143 from PathwayCommons/imgspd

Add controls for max. number of images to generate in parallel

Author: jvwong

README.md

@@ -97,6 +97,8 @@ Demo:
 
 Sharing:
 
+- `DOCUMENT_IMAGE_CACHE_SIZE` : number of images to cache in memory
+- `DOCUMENT_IMAGE_PLL_LIMIT` : max number of images to be generated in parallel (expensive)
 - `DOCUMENT_IMAGE_WIDTH` : tweet card image width
 - `DOCUMENT_IMAGE_HEIGHT` : tweet card image height
 - `DOCUMENT_IMAGE_PADDING` : padding around tweet card image (prevents twitter cropping issues)

src/config.js

@@ -110,6 +110,7 @@ export const EMAIL_ADDRESS_ADMIN = env('EMAIL_ADDRESS_ADMIN', 'pc@biofactoid.com
 
 // Sharing
 export const DOCUMENT_IMAGE_CACHE_SIZE = env('DOCUMENT_IMAGE_CACHE_SIZE', 500);
+export const DOCUMENT_IMAGE_PLL_LIMIT = env('DOCUMENT_IMAGE_PLL_LIMIT', 1);
 export const DOCUMENT_IMAGE_WIDTH = env('DOCUMENT_IMAGE_WIDTH', 2400);
 export const DOCUMENT_IMAGE_HEIGHT = env('DOCUMENT_IMAGE_HEIGHT', 1200);
 export const DOCUMENT_IMAGE_PADDING = env('DOCUMENT_IMAGE_PADDING', 50);

src/server/routes/api/document/index.js

@@ -11,6 +11,7 @@ import emailRegex from 'email-regex';
 import url from 'url';
 import { URLSearchParams } from  'url';
 import NodeCache from 'node-cache';
+import pLimit from './p-limit';
 
 import { tryPromise, makeStaticStylesheet, makeCyEles, truncateString } from '../../../../util';
 import { msgFactory, updateCorrespondence, EmailError } from '../../../email';
@@ -54,9 +55,12 @@ import { BASE_URL,
   BIOPAX_DOWNLOADS_PATH,
   BIOPAX_IDMAP_DOWNLOADS_PATH,
   ORCID_PUBLIC_API_BASE_URL,
-  DOI_LINK_BASE_URL
+  DOI_LINK_BASE_URL,
+  DOCUMENT_IMAGE_PLL_LIMIT
  } from '../../../../config';
 
+
+
 import { ENTITY_TYPE } from '../../../../model/element/entity-type';
 import { eLink, elink2UidList } from './pubmed/linkPubmed';
 import { fetchPubmed } from './pubmed/fetchPubmed';
@@ -1407,6 +1411,12 @@ const getDocumentImageBuffer = doc => {
   );
 };
 
+const imgLimit = pLimit(DOCUMENT_IMAGE_PLL_LIMIT);
+
+const getDocumentImageBufferRatedLimited = doc => {
+  return imgLimit(() => getDocumentImageBuffer(doc));
+};
+
 const imageCache = new LRUCache({
   max: DOCUMENT_IMAGE_CACHE_SIZE
 });
@@ -1448,7 +1458,7 @@ http.get('/(:id).png', function( req, res, next ){
   };
 
   const fillCache = async doc => {
-    const img = await getDocumentImageBuffer(doc);
+    const img = await getDocumentImageBufferRatedLimited(doc);
     const cache = { img };
     const ttl = calcTtl(doc);
 

src/server/routes/api/document/p-limit.js

@@ -0,0 +1,151 @@
+/**
+ * This file is jury-rigged.  It's the entire transitive code of 'p-limit'.
+ * 
+ * - p-limit does not support CJS.
+ * - esm does not support p-limit.
+ * - This could be resolved in future by upgrading the entire project to use built-in ESM
+ *   in new versions of node.  However, all the imports would need to be changed.  It would
+ *   be a major project.
+ */
+
+
+
+
+
+
+
+
+/*
+How it works:
+`this.#head` is an instance of `Node` which keeps track of its current value and nests another instance of `Node` that keeps the value that comes after it. When a value is provided to `.enqueue()`, the code needs to iterate through `this.#head`, going deeper and deeper to find the last value. However, iterating through every single item is slow. This problem is solved by saving a reference to the last value as `this.#tail` so that it can reference it to add a new value.
+*/
+
+class Node {
+	value;
+	next;
+
+	constructor(value) {
+		this.value = value;
+	}
+}
+
+class Queue {
+	#head;
+	#tail;
+	#size;
+
+	constructor() {
+		this.clear();
+	}
+
+	enqueue(value) {
+		const node = new Node(value);
+
+		if (this.#head) {
+			this.#tail.next = node;
+			this.#tail = node;
+		} else {
+			this.#head = node;
+			this.#tail = node;
+		}
+
+		this.#size++;
+	}
+
+	dequeue() {
+		const current = this.#head;
+		if (!current) {
+			return;
+		}
+
+		this.#head = this.#head.next;
+		this.#size--;
+		return current.value;
+	}
+
+	clear() {
+		this.#head = undefined;
+		this.#tail = undefined;
+		this.#size = 0;
+	}
+
+	get size() {
+		return this.#size;
+	}
+
+	* [Symbol.iterator]() {
+		let current = this.#head;
+
+		while (current) {
+			yield current.value;
+			current = current.next;
+		}
+	}
+}
+
+export default function pLimit(concurrency) {
+	if (!((Number.isInteger(concurrency) || concurrency === Number.POSITIVE_INFINITY) && concurrency > 0)) {
+		throw new TypeError('Expected `concurrency` to be a number from 1 and up');
+	}
+
+	const queue = new Queue();
+	let activeCount = 0;
+
+	const next = () => {
+		activeCount--;
+
+		if (queue.size > 0) {
+			queue.dequeue()();
+		}
+	};
+
+	const run = async (fn, resolve, args) => {
+		activeCount++;
+
+		const result = (async () => fn(...args))();
+
+		resolve(result);
+
+		try {
+			await result;
+		} catch {}
+
+		next();
+	};
+
+	const enqueue = (fn, resolve, args) => {
+		queue.enqueue(run.bind(undefined, fn, resolve, args));
+
+		(async () => {
+			// This function needs to wait until the next microtask before comparing
+			// `activeCount` to `concurrency`, because `activeCount` is updated asynchronously
+			// when the run function is dequeued and called. The comparison in the if-statement
+			// needs to happen asynchronously as well to get an up-to-date value for `activeCount`.
+			await Promise.resolve();
+
+			if (activeCount < concurrency && queue.size > 0) {
+				queue.dequeue()();
+			}
+		})();
+	};
+
+	const generator = (fn, ...args) => new Promise(resolve => {
+		enqueue(fn, resolve, args);
+	});
+
+	Object.defineProperties(generator, {
+		activeCount: {
+			get: () => activeCount,
+		},
+		pendingCount: {
+			get: () => queue.size,
+		},
+		clearQueue: {
+			value: () => {
+				queue.clear();
+			},
+		},
+	});
+
+	return generator;
+}