<NME> AsyncQueue.js
<BEF> /*
	MIT License http://www.opensource.org/licenses/mit-license.php
	Author Tobias Koppers @sokra
*/

"use strict";

const { SyncHook, AsyncSeriesHook } = require("tapable");
const { makeWebpackError } = require("../HookWebpackError");
const WebpackError = require("../WebpackError");

/**
 * @template T
 * @template R
 */
class AsyncQueue {

	 * @param {Object} options options object
	 * @param {string=} options.name name of the queue
	 * @param {number} options.parallelism how many items should be processed at once
	 * @param {function(T, Callback<R>): void} options.processor async function to process items
	 */
	constructor({ name, parallelism, processor }) {
		this._name = name;
		this._parallelism = parallelism;
		this._processor = processor;
		/** @type {Map<T, Callback<R>[]>} */
		this._callbacks = new Map();
		/** @type {Set<T>} */
		this._queued = new Set();
		/** @type {Set<T>} */
		this._processing = new Set();
		/** @type {Map<T, [Error, R]>} */
		this._results = new Map();
		this._activeTasks = 0;
		this._willEnsureProcessing = false;
		/** @type {Callback<R>[] | undefined} */
		this.callbacks = undefined;
		this.result = undefined;
		/** @type {WebpackError | undefined} */
		this.error = undefined;
	}
}

/**
 * @template T
 * @template K
 * @template R
 */
class AsyncQueue {
	/**
	 * @param {Object} options options object
	 * @param {string=} options.name name of the queue
	 * @param {number=} options.parallelism how many items should be processed at once
	 * @param {AsyncQueue<any, any, any>=} options.parent parent queue, which will have priority over this queue and with shared parallelism
	 * @param {function(T): K=} options.getKey extract key from item
	 * @param {function(T, Callback<R>): void} options.processor async function to process items
	 */
				callback(err);
				return;
			}
			const result = this._results.get(item);
			if (result !== undefined) {
				process.nextTick(() => callback(result[0], result[1]));
				return;
			}
			let callbacks = this._callbacks.get(item);
			if (callbacks !== undefined) {
				callbacks.push(callback);
				return;
			}
			callbacks = [callback];
			this._callbacks.set(item, callbacks);
			if (this._stopped) {
				this.hooks.added.call(item);
				this._activeTasks++;
				this._root._children = [this];
			} else {
				this._root._children.push(this);
			}
		}

		this.hooks = {
			/** @type {AsyncSeriesHook<[T]>} */
			beforeAdd: new AsyncSeriesHook(["item"]),
			/** @type {SyncHook<[T]>} */
			added: new SyncHook(["item"]),
			/** @type {AsyncSeriesHook<[T]>} */
			beforeStart: new AsyncSeriesHook(["item"]),
			/** @type {SyncHook<[T]>} */
	 * @returns {void}
	 */
	invalidate(item) {
		this._results.delete(item);
	}

	/**
	/**
	 * @param {T} item an item
	 * @param {Callback<R>} callback callback function
	 * @returns {void}
	 */
	add(item, callback) {
		if (this._stopped) return callback(new WebpackError("Queue was stopped"));
		this.hooks.beforeAdd.callAsync(item, err => {
			if (err) {
				callback(
					makeWebpackError(err, `AsyncQueue(${this._name}).hooks.beforeAdd`)
				);
				return;
			}
			const key = this._getKey(item);
			const entry = this._entries.get(key);
			if (entry !== undefined) {
				if (entry.state === DONE_STATE) {
					if (inHandleResult++ > 3) {
						process.nextTick(() => callback(entry.error, entry.result));
					} else {
						callback(entry.error, entry.result);
					}
					inHandleResult--;
				} else if (entry.callbacks === undefined) {
					entry.callbacks = [callback];
				} else {
					entry.callbacks.push(callback);
				}
				return;
			}
	 * @returns {boolean} true, if the item is currently being processed
	 */
	isProcessing(item) {
		return this._processing.has(item);
	}

	/**
				this._entries.set(key, newEntry);
				this._queued.enqueue(newEntry);
				const root = this._root;
				root._needProcessing = true;
				if (root._willEnsureProcessing === false) {
					root._willEnsureProcessing = true;
					setImmediate(root._ensureProcessing);
				}
				this.hooks.added.call(item);
	 * @returns {boolean} true, if the item is currently queued
	 */
	isDone(item) {
		return this._results.has(item);
	}

	/**
	invalidate(item) {
		const key = this._getKey(item);
		const entry = this._entries.get(key);
		this._entries.delete(key);
		if (entry.state === QUEUED_STATE) {
			this._queued.delete(entry);
		}
		for (const item of this._queued) {
			this._activeTasks++;
			this._queued.delete(item);
			this._processing.add(item);
			this._startProcessing(item);
			if (this._activeTasks >= this._parallelism) {
				this._willEnsureProcessing = false;
	waitFor(item, callback) {
		const key = this._getKey(item);
		const entry = this._entries.get(key);
		if (entry === undefined) {
			return callback(
				new WebpackError(
					"waitFor can only be called for an already started item"
				)
			);
		}
		if (entry.state === DONE_STATE) {
			process.nextTick(() => callback(entry.error, entry.result));
		} else if (entry.callbacks === undefined) {
			entry.callbacks = [callback];
		} else {
			entry.callbacks.push(callback);
		}
	}

	/**
	 * @returns {void}
	 */
	stop() {
		this._stopped = true;
		const queue = this._queued;
		this._queued = new ArrayQueue();
		const root = this._root;
		for (const entry of queue) {
			this._entries.delete(this._getKey(entry.item));
			root._activeTasks++;
			this._handleResult(entry, new WebpackError("Queue was stopped"));
		}
	}

	/**
	 * @returns {void}
	 */
		this.hooks.result.callAsync(item, err, result, hookError => {
			const error = hookError || err;

			const callbacks = this._callbacks.get(item);
			this._processing.delete(item);
			this._results.set(item, [error, result]);
			this._callbacks.delete(item);
			this._activeTasks--;

			if (this._willEnsureProcessing === false && this._queued.size > 0) {
	 * @returns {void}
	 */
	decreaseParallelism() {
		const root = this._root;
		root._parallelism--;
	}

	/**
	 * @param {T} item an item
	 * @returns {boolean} true, if the item is currently being processed
	 */
	isProcessing(item) {
		const key = this._getKey(item);
		const entry = this._entries.get(key);
		return entry !== undefined && entry.state === PROCESSING_STATE;
	}

	/**
	 * @param {T} item an item
	 * @returns {boolean} true, if the item is currently queued
	 */
	isQueued(item) {
		const key = this._getKey(item);
		const entry = this._entries.get(key);
		return entry !== undefined && entry.state === QUEUED_STATE;
	}

	/**
	 * @param {T} item an item
	 * @returns {boolean} true, if the item is currently queued
	 */
	isDone(item) {
		const key = this._getKey(item);
		const entry = this._entries.get(key);
		return entry !== undefined && entry.state === DONE_STATE;
	}

	/**
	 * @returns {void}
	 */
	_ensureProcessing() {
		while (this._activeTasks < this._parallelism) {
			const entry = this._queued.dequeue();
			if (entry === undefined) break;
			this._activeTasks++;
			entry.state = PROCESSING_STATE;
			this._startProcessing(entry);
		}
		this._willEnsureProcessing = false;
		if (this._queued.length > 0) return;
		if (this._children !== undefined) {
			for (const child of this._children) {
				while (this._activeTasks < this._parallelism) {
					const entry = child._queued.dequeue();
					if (entry === undefined) break;
					this._activeTasks++;
					entry.state = PROCESSING_STATE;
					child._startProcessing(entry);
				}
				if (child._queued.length > 0) return;
			}
		}
		if (!this._willEnsureProcessing) this._needProcessing = false;
	}

	/**
	 * @param {AsyncQueueEntry<T, K, R>} entry the entry
	 * @returns {void}
	 */
	_startProcessing(entry) {
		this.hooks.beforeStart.callAsync(entry.item, err => {
			if (err) {
				this._handleResult(
					entry,
					makeWebpackError(err, `AsyncQueue(${this._name}).hooks.beforeStart`)
				);
				return;
			}
			let inCallback = false;
			try {
				this._processor(entry.item, (e, r) => {
					inCallback = true;
					this._handleResult(entry, e, r);
				});
			} catch (err) {
				if (inCallback) throw err;
				this._handleResult(entry, err, null);
			}
			this.hooks.started.call(entry.item);
		});
	}

	/**
	 * @param {AsyncQueueEntry<T, K, R>} entry the entry
	 * @param {WebpackError=} err error, if any
	 * @param {R=} result result, if any
	 * @returns {void}
	 */
	_handleResult(entry, err, result) {
		this.hooks.result.callAsync(entry.item, err, result, hookError => {
			const error = hookError
				? makeWebpackError(hookError, `AsyncQueue(${this._name}).hooks.result`)
				: err;

			const callback = entry.callback;
			const callbacks = entry.callbacks;
			entry.state = DONE_STATE;
			entry.callback = undefined;
			entry.callbacks = undefined;
			entry.result = result;
			entry.error = error;

			const root = this._root;
			root._activeTasks--;
			if (root._willEnsureProcessing === false && root._needProcessing) {
				root._willEnsureProcessing = true;
				setImmediate(root._ensureProcessing);
			}

			if (inHandleResult++ > 3) {
				process.nextTick(() => {
					callback(error, result);
					if (callbacks !== undefined) {
						for (const callback of callbacks) {
							callback(error, result);
						}
					}
				});
			} else {
				callback(error, result);
				if (callbacks !== undefined) {
					for (const callback of callbacks) {
						callback(error, result);
					}
				}
			}
			inHandleResult--;
		});
	}

	clear() {
		this._entries.clear();
		this._queued.clear();
		this._activeTasks = 0;
		this._willEnsureProcessing = false;
		this._needProcessing = false;
		this._stopped = false;
	}
}

module.exports = AsyncQueue;

<MSG> make Compilation.addModule async and queued

<DFF> @@ -11,6 +11,7 @@ const { SyncHook, AsyncSeriesHook } = require("tapable");
 
 /**
  * @template T
+ * @template K
  * @template R
  */
 class AsyncQueue {
@@ -18,19 +19,22 @@ class AsyncQueue {
 	 * @param {Object} options options object
 	 * @param {string=} options.name name of the queue
 	 * @param {number} options.parallelism how many items should be processed at once
+	 * @param {function(T): K=} options.getKey extract key from item
 	 * @param {function(T, Callback<R>): void} options.processor async function to process items
 	 */
-	constructor({ name, parallelism, processor }) {
+	constructor({ name, parallelism, processor, getKey }) {
 		this._name = name;
 		this._parallelism = parallelism;
 		this._processor = processor;
-		/** @type {Map<T, Callback<R>[]>} */
+		this._getKey =
+			getKey || /** @type {(T) => K} */ (item => /** @type {any} */ (item));
+		/** @type {Map<K, Callback<R>[]>} */
 		this._callbacks = new Map();
 		/** @type {Set<T>} */
 		this._queued = new Set();
-		/** @type {Set<T>} */
+		/** @type {Set<K>} */
 		this._processing = new Set();
-		/** @type {Map<T, [Error, R]>} */
+		/** @type {Map<K, [Error, R]>} */
 		this._results = new Map();
 		this._activeTasks = 0;
 		this._willEnsureProcessing = false;
@@ -59,18 +63,19 @@ class AsyncQueue {
 				callback(err);
 				return;
 			}
-			const result = this._results.get(item);
+			const key = this._getKey(item);
+			const result = this._results.get(key);
 			if (result !== undefined) {
 				process.nextTick(() => callback(result[0], result[1]));
 				return;
 			}
-			let callbacks = this._callbacks.get(item);
+			let callbacks = this._callbacks.get(key);
 			if (callbacks !== undefined) {
 				callbacks.push(callback);
 				return;
 			}
 			callbacks = [callback];
-			this._callbacks.set(item, callbacks);
+			this._callbacks.set(key, callbacks);
 			if (this._stopped) {
 				this.hooks.added.call(item);
 				this._activeTasks++;
@@ -91,7 +96,8 @@ class AsyncQueue {
 	 * @returns {void}
 	 */
 	invalidate(item) {
-		this._results.delete(item);
+		const key = this._getKey(item);
+		this._results.delete(key);
 	}
 
 	/**
@@ -129,7 +135,8 @@ class AsyncQueue {
 	 * @returns {boolean} true, if the item is currently being processed
 	 */
 	isProcessing(item) {
-		return this._processing.has(item);
+		const key = this._getKey(item);
+		return this._processing.has(key);
 	}
 
 	/**
@@ -145,7 +152,8 @@ class AsyncQueue {
 	 * @returns {boolean} true, if the item is currently queued
 	 */
 	isDone(item) {
-		return this._results.has(item);
+		const key = this._getKey(item);
+		return this._results.has(key);
 	}
 
 	/**
@@ -158,8 +166,9 @@ class AsyncQueue {
 		}
 		for (const item of this._queued) {
 			this._activeTasks++;
+			const key = this._getKey(item);
 			this._queued.delete(item);
-			this._processing.add(item);
+			this._processing.add(key);
 			this._startProcessing(item);
 			if (this._activeTasks >= this._parallelism) {
 				this._willEnsureProcessing = false;
@@ -203,10 +212,11 @@ class AsyncQueue {
 		this.hooks.result.callAsync(item, err, result, hookError => {
 			const error = hookError || err;
 
-			const callbacks = this._callbacks.get(item);
-			this._processing.delete(item);
-			this._results.set(item, [error, result]);
-			this._callbacks.delete(item);
+			const key = this._getKey(item);
+			const callbacks = this._callbacks.get(key);
+			this._processing.delete(key);
+			this._results.set(key, [error, result]);
+			this._callbacks.delete(key);
 			this._activeTasks--;
 
 			if (this._willEnsureProcessing === false && this._queued.size > 0) {

{'commit': 'diff --git a/lib/Compilation.js b/lib/Compilation.js\nindex 9326a2dbc..21053f3aa 100644\n--- a/lib/Compilation.js\n+++ b/lib/Compilation.js\n@@ -375,21 +375,32 @@ class Compilation {\n \t\tthis.moduleGraph = new ModuleGraph();\n \t\tthis.chunkGraph = undefined;\n \n+\t\t/** @type {AsyncQueue<TODO, TODO, Module>} */\n \t\tthis.factorizeQueue = new AsyncQueue({\n \t\t\tname: "factorize",\n \t\t\tparallelism: options.parallelism || 100,\n \t\t\tprocessor: this._factorizeModule.bind(this)\n \t\t});\n+\t\t/** @type {AsyncQueue<Module, string, Module>} */\n+\t\tthis.addModuleQueue = new AsyncQueue({\n+\t\t\tname: "addModule",\n+\t\t\tparallelism: options.parallelism || 100,\n+\t\t\tgetKey: module => module.identifier(),\n+\t\t\tprocessor: this._addModule.bind(this)\n+\t\t});\n+\t\t/** @type {AsyncQueue<Module, Module, Module>} */\n \t\tthis.buildQueue = new AsyncQueue({\n \t\t\tname: "build",\n \t\t\tparallelism: options.parallelism || 100,\n \t\t\tprocessor: this._buildModule.bind(this)\n \t\t});\n+\t\t/** @type {AsyncQueue<Module, Module, Module>} */\n \t\tthis.rebuildQueue = new AsyncQueue({\n \t\t\tname: "rebuild",\n \t\t\tparallelism: options.parallelism || 100,\n \t\t\tprocessor: this._rebuildModule.bind(this)\n \t\t});\n+\t\t/** @type {AsyncQueue<Module, Module, Module>} */\n \t\tthis.processDependenciesQueue = new AsyncQueue({\n \t\t\tname: "processDependencies",\n \t\t\tparallelism: options.parallelism || 100,\n@@ -461,14 +472,25 @@ class Compilation {\n \n \t/**\n \t * @param {Module} module module to be added that was created\n-\t * @returns {Module} returns the module in the compilation,\n+\t * @param {ModuleCallback} callback returns the module in the compilation,\n \t * it could be the passed one (if new), or an already existing in the compilation\n+\t * @returns {void}\n \t */\n-\taddModule(module) {\n+\taddModule(module, callback) {\n+\t\tthis.addModuleQueue.add(module, callback);\n+\t}\n+\n+\t/**\n+\t * @param {Module} module module to be added that was created\n+\t * @param {ModuleCallback} callback returns the module in the compilation,\n+\t * it could be the passed one (if new), or an already existing in the compilation\n+\t * @returns {void}\n+\t */\n+\t_addModule(module, callback) {\n \t\tconst identifier = module.identifier();\n \t\tconst alreadyAddedModule = this._modules.get(identifier);\n \t\tif (alreadyAddedModule) {\n-\t\t\treturn alreadyAddedModule;\n+\t\t\treturn callback(null, alreadyAddedModule);\n \t\t}\n \t\tconst cacheName = "m" + identifier;\n \t\tif (this.cache && this.cache[cacheName]) {\n@@ -477,14 +499,15 @@ class Compilation {\n \t\t\tcacheModule.updateCacheModule(module);\n \n \t\t\tmodule = cacheModule;\n+\t\t} else {\n+\t\t\tif (this.cache) {\n+\t\t\t\tthis.cache[cacheName] = module;\n+\t\t\t}\n \t\t}\n \t\tthis._modules.set(identifier, module);\n-\t\tif (this.cache) {\n-\t\t\tthis.cache[cacheName] = module;\n-\t\t}\n \t\tthis.modules.add(module);\n \t\tModuleGraph.setModuleGraphForModule(module, this.moduleGraph);\n-\t\treturn module;\n+\t\treturn callback(null, module);\n \t}\n \n \t/**\n@@ -714,30 +737,8 @@ class Compilation {\n \t\t\t\t\treturn callback();\n \t\t\t\t}\n \n-\t\t\t\tconst module = this.addModule(newModule);\n-\n-\t\t\t\tfor (let i = 0; i < dependencies.length; i++) {\n-\t\t\t\t\tconst dependency = dependencies[i];\n-\t\t\t\t\tmoduleGraph.setResolvedModule(originModule, dependency, module);\n-\t\t\t\t}\n-\n-\t\t\t\tif (module === newModule) {\n-\t\t\t\t\tif (currentProfile !== undefined) {\n-\t\t\t\t\t\tmoduleGraph.setProfile(module, currentProfile);\n-\t\t\t\t\t}\n-\n-\t\t\t\t\tif (originModule !== undefined) {\n-\t\t\t\t\t\tmoduleGraph.setIssuer(module, originModule);\n-\t\t\t\t\t}\n-\t\t\t\t} else {\n-\t\t\t\t\tif (currentProfile !== undefined) {\n-\t\t\t\t\t\tcurrentProfile.mergeInto(moduleGraph.getProfile(module));\n-\t\t\t\t\t}\n-\t\t\t\t}\n-\n-\t\t\t\tthis.buildModule(module, error => {\n-\t\t\t\t\tif (error) {\n-\t\t\t\t\t\tconst err = /** @type {WebpackError} */ (error);\n+\t\t\t\tthis.addModule(newModule, (err, module) => {\n+\t\t\t\t\tif (err) {\n \t\t\t\t\t\tif (!err.module) {\n \t\t\t\t\t\t\terr.module = module;\n \t\t\t\t\t\t}\n@@ -746,16 +747,46 @@ class Compilation {\n \t\t\t\t\t\treturn callback(err);\n \t\t\t\t\t}\n \n-\t\t\t\t\t// This avoids deadlocks for circular dependencies\n-\t\t\t\t\tif (this.processDependenciesQueue.isProcessing(module)) {\n-\t\t\t\t\t\treturn callback();\n+\t\t\t\t\tfor (let i = 0; i < dependencies.length; i++) {\n+\t\t\t\t\t\tconst dependency = dependencies[i];\n+\t\t\t\t\t\tmoduleGraph.setResolvedModule(originModule, dependency, module);\n \t\t\t\t\t}\n \n-\t\t\t\t\tthis.processModuleDependencies(module, err => {\n+\t\t\t\t\tif (module === newModule) {\n+\t\t\t\t\t\tif (currentProfile !== undefined) {\n+\t\t\t\t\t\t\tmoduleGraph.setProfile(module, currentProfile);\n+\t\t\t\t\t\t}\n+\n+\t\t\t\t\t\tif (originModule !== undefined) {\n+\t\t\t\t\t\t\tmoduleGraph.setIssuer(module, originModule);\n+\t\t\t\t\t\t}\n+\t\t\t\t\t} else {\n+\t\t\t\t\t\tif (currentProfile !== undefined) {\n+\t\t\t\t\t\t\tcurrentProfile.mergeInto(moduleGraph.getProfile(module));\n+\t\t\t\t\t\t}\n+\t\t\t\t\t}\n+\n+\t\t\t\t\tthis.buildModule(module, err => {\n \t\t\t\t\t\tif (err) {\n+\t\t\t\t\t\t\tif (!err.module) {\n+\t\t\t\t\t\t\t\terr.module = module;\n+\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\tthis.errors.push(err);\n+\n \t\t\t\t\t\t\treturn callback(err);\n \t\t\t\t\t\t}\n-\t\t\t\t\t\tcallback(null, module);\n+\n+\t\t\t\t\t\t// This avoids deadlocks for circular dependencies\n+\t\t\t\t\t\tif (this.processDependenciesQueue.isProcessing(module)) {\n+\t\t\t\t\t\t\treturn callback();\n+\t\t\t\t\t\t}\n+\n+\t\t\t\t\t\tthis.processModuleDependencies(module, err => {\n+\t\t\t\t\t\t\tif (err) {\n+\t\t\t\t\t\t\t\treturn callback(err);\n+\t\t\t\t\t\t\t}\n+\t\t\t\t\t\t\tcallback(null, module);\n+\t\t\t\t\t\t});\n \t\t\t\t\t});\n \t\t\t\t});\n \t\t\t}\ndiff --git a/lib/util/AsyncQueue.js b/lib/util/AsyncQueue.js\nindex c8d749359..24dc09cc7 100644\n--- a/lib/util/AsyncQueue.js\n+++ b/lib/util/AsyncQueue.js\n@@ -11,6 +11,7 @@ const { SyncHook, AsyncSeriesHook } = require("tapable");\n \n /**\n  * @template T\n+ * @template K\n  * @template R\n  */\n class AsyncQueue {\n@@ -18,19 +19,22 @@ class AsyncQueue {\n \t * @param {Object} options options object\n \t * @param {string=} options.name name of the queue\n \t * @param {number} options.parallelism how many items should be processed at once\n+\t * @param {function(T): K=} options.getKey extract key from item\n \t * @param {function(T, Callback<R>): void} options.processor async function to process items\n \t */\n-\tconstructor({ name, parallelism, processor }) {\n+\tconstructor({ name, parallelism, processor, getKey }) {\n \t\tthis._name = name;\n \t\tthis._parallelism = parallelism;\n \t\tthis._processor = processor;\n-\t\t/** @type {Map<T, Callback<R>[]>} */\n+\t\tthis._getKey =\n+\t\t\tgetKey || /** @type {(T) => K} */ (item => /** @type {any} */ (item));\n+\t\t/** @type {Map<K, Callback<R>[]>} */\n \t\tthis._callbacks = new Map();\n \t\t/** @type {Set<T>} */\n \t\tthis._queued = new Set();\n-\t\t/** @type {Set<T>} */\n+\t\t/** @type {Set<K>} */\n \t\tthis._processing = new Set();\n-\t\t/** @type {Map<T, [Error, R]>} */\n+\t\t/** @type {Map<K, [Error, R]>} */\n \t\tthis._results = new Map();\n \t\tthis._activeTasks = 0;\n \t\tthis._willEnsureProcessing = false;\n@@ -59,18 +63,19 @@ class AsyncQueue {\n \t\t\t\tcallback(err);\n \t\t\t\treturn;\n \t\t\t}\n-\t\t\tconst result = this._results.get(item);\n+\t\t\tconst key = this._getKey(item);\n+\t\t\tconst result = this._results.get(key);\n \t\t\tif (result !== undefined) {\n \t\t\t\tprocess.nextTick(() => callback(result[0], result[1]));\n \t\t\t\treturn;\n \t\t\t}\n-\t\t\tlet callbacks = this._callbacks.get(item);\n+\t\t\tlet callbacks = this._callbacks.get(key);\n \t\t\tif (callbacks !== undefined) {\n \t\t\t\tcallbacks.push(callback);\n \t\t\t\treturn;\n \t\t\t}\n \t\t\tcallbacks = [callback];\n-\t\t\tthis._callbacks.set(item, callbacks);\n+\t\t\tthis._callbacks.set(key, callbacks);\n \t\t\tif (this._stopped) {\n \t\t\t\tthis.hooks.added.call(item);\n \t\t\t\tthis._activeTasks++;\n@@ -91,7 +96,8 @@ class AsyncQueue {\n \t * @returns {void}\n \t */\n \tinvalidate(item) {\n-\t\tthis._results.delete(item);\n+\t\tconst key = this._getKey(item);\n+\t\tthis._results.delete(key);\n \t}\n \n \t/**\n@@ -129,7 +135,8 @@ class AsyncQueue {\n \t * @returns {boolean} true, if the item is currently being processed\n \t */\n \tisProcessing(item) {\n-\t\treturn this._processing.has(item);\n+\t\tconst key = this._getKey(item);\n+\t\treturn this._processing.has(key);\n \t}\n \n \t/**\n@@ -145,7 +152,8 @@ class AsyncQueue {\n \t * @returns {boolean} true, if the item is currently queued\n \t */\n \tisDone(item) {\n-\t\treturn this._results.has(item);\n+\t\tconst key = this._getKey(item);\n+\t\treturn this._results.has(key);\n \t}\n \n \t/**\n@@ -158,8 +166,9 @@ class AsyncQueue {\n \t\t}\n \t\tfor (const item of this._queued) {\n \t\t\tthis._activeTasks++;\n+\t\t\tconst key = this._getKey(item);\n \t\t\tthis._queued.delete(item);\n-\t\t\tthis._processing.add(item);\n+\t\t\tthis._processing.add(key);\n \t\t\tthis._startProcessing(item);\n \t\t\tif (this._activeTasks >= this._parallelism) {\n \t\t\t\tthis._willEnsureProcessing = false;\n@@ -203,10 +212,11 @@ class AsyncQueue {\n \t\tthis.hooks.result.callAsync(item, err, result, hookError => {\n \t\t\tconst error = hookError || err;\n \n-\t\t\tconst callbacks = this._callbacks.get(item);\n-\t\t\tthis._processing.delete(item);\n-\t\t\tthis._results.set(item, [error, result]);\n-\t\t\tthis._callbacks.delete(item);\n+\t\t\tconst key = this._getKey(item);\n+\t\t\tconst callbacks = this._callbacks.get(key);\n+\t\t\tthis._processing.delete(key);\n+\t\t\tthis._results.set(key, [error, result]);\n+\t\t\tthis._callbacks.delete(key);\n \t\t\tthis._activeTasks--;\n \n \t\t\tif (this._willEnsureProcessing === false && this._queued.size > 0) {\n', 'commit_subject': 'make Compilation.addModule async and queued', 'repo_name': 'webpack/webpack', 'committer': {'name': 'Tobias Koppers', 'email': '0cad4b845df78a4c17cdaf2dcf19cc8e39488037@googlemail.com', 'time_sec': '1538040071', 'tz_offset': '120', 'date': {'seconds': '1538040071'}}, 'commit_tree': 'ecaf067b3a35c8e2c0b55099d7cfea4070a15526', 'commit_message': 'make Compilation.addModule async and queued\n', 'language_name': 'JavaScript', 'license': 'mit', 'hunks': '@@ -11,6 +11,7 @@ const { SyncHook, AsyncSeriesHook } = require("tapable");\n \n /**\n  * @template T\n+ * @template K\n  * @template R\n  */\n class AsyncQueue {\n@@ -18,19 +19,22 @@ class AsyncQueue {\n \t * @param {Object} options options object\n \t * @param {string=} options.name name of the queue\n \t * @param {number} options.parallelism how many items should be processed at once\n+\t * @param {function(T): K=} options.getKey extract key from item\n \t * @param {function(T, Callback<R>): void} options.processor async function to process items\n \t */\n-\tconstructor({ name, parallelism, processor }) {\n+\tconstructor({ name, parallelism, processor, getKey }) {\n \t\tthis._name = name;\n \t\tthis._parallelism = parallelism;\n \t\tthis._processor = processor;\n-\t\t/** @type {Map<T, Callback<R>[]>} */\n+\t\tthis._getKey =\n+\t\t\tgetKey || /** @type {(T) => K} */ (item => /** @type {any} */ (item));\n+\t\t/** @type {Map<K, Callback<R>[]>} */\n \t\tthis._callbacks = new Map();\n \t\t/** @type {Set<T>} */\n \t\tthis._queued = new Set();\n-\t\t/** @type {Set<T>} */\n+\t\t/** @type {Set<K>} */\n \t\tthis._processing = new Set();\n-\t\t/** @type {Map<T, [Error, R]>} */\n+\t\t/** @type {Map<K, [Error, R]>} */\n \t\tthis._results = new Map();\n \t\tthis._activeTasks = 0;\n \t\tthis._willEnsureProcessing = false;\n@@ -59,18 +63,19 @@ class AsyncQueue {\n \t\t\t\tcallback(err);\n \t\t\t\treturn;\n \t\t\t}\n-\t\t\tconst result = this._results.get(item);\n+\t\t\tconst key = this._getKey(item);\n+\t\t\tconst result = this._results.get(key);\n \t\t\tif (result !== undefined) {\n \t\t\t\tprocess.nextTick(() => callback(result[0], result[1]));\n \t\t\t\treturn;\n \t\t\t}\n-\t\t\tlet callbacks = this._callbacks.get(item);\n+\t\t\tlet callbacks = this._callbacks.get(key);\n \t\t\tif (callbacks !== undefined) {\n \t\t\t\tcallbacks.push(callback);\n \t\t\t\treturn;\n \t\t\t}\n \t\t\tcallbacks = [callback];\n-\t\t\tthis._callbacks.set(item, callbacks);\n+\t\t\tthis._callbacks.set(key, callbacks);\n \t\t\tif (this._stopped) {\n \t\t\t\tthis.hooks.added.call(item);\n \t\t\t\tthis._activeTasks++;\n@@ -91,7 +96,8 @@ class AsyncQueue {\n \t * @returns {void}\n \t */\n \tinvalidate(item) {\n-\t\tthis._results.delete(item);\n+\t\tconst key = this._getKey(item);\n+\t\tthis._results.delete(key);\n \t}\n \n \t/**\n@@ -129,7 +135,8 @@ class AsyncQueue {\n \t * @returns {boolean} true, if the item is currently being processed\n \t */\n \tisProcessing(item) {\n-\t\treturn this._processing.has(item);\n+\t\tconst key = this._getKey(item);\n+\t\treturn this._processing.has(key);\n \t}\n \n \t/**\n@@ -145,7 +152,8 @@ class AsyncQueue {\n \t * @returns {boolean} true, if the item is currently queued\n \t */\n \tisDone(item) {\n-\t\treturn this._results.has(item);\n+\t\tconst key = this._getKey(item);\n+\t\treturn this._results.has(key);\n \t}\n \n \t/**\n@@ -158,8 +166,9 @@ class AsyncQueue {\n \t\t}\n \t\tfor (const item of this._queued) {\n \t\t\tthis._activeTasks++;\n+\t\t\tconst key = this._getKey(item);\n \t\t\tthis._queued.delete(item);\n-\t\t\tthis._processing.add(item);\n+\t\t\tthis._processing.add(key);\n \t\t\tthis._startProcessing(item);\n \t\t\tif (this._activeTasks >= this._parallelism) {\n \t\t\t\tthis._willEnsureProcessing = false;\n@@ -203,10 +212,11 @@ class AsyncQueue {\n \t\tthis.hooks.result.callAsync(item, err, result, hookError => {\n \t\t\tconst error = hookError || err;\n \n-\t\t\tconst callbacks = this._callbacks.get(item);\n-\t\t\tthis._processing.delete(item);\n-\t\t\tthis._results.set(item, [error, result]);\n-\t\t\tthis._callbacks.delete(item);\n+\t\t\tconst key = this._getKey(item);\n+\t\t\tconst callbacks = this._callbacks.get(key);\n+\t\t\tthis._processing.delete(key);\n+\t\t\tthis._results.set(key, [error, result]);\n+\t\t\tthis._callbacks.delete(key);\n \t\t\tthis._activeTasks--;\n \n \t\t\tif (this._willEnsureProcessing === false && this._queued.size > 0) {\n', 'addition_count': 25, 'deletion_count': 15, 'src_file': 'a/lib/util/AsyncQueue.js', 'tgt_file': 'b/lib/util/AsyncQueue.js', 'file_extension': '.js', 'before_file': '/*\n\tMIT License http://www.opensource.org/licenses/mit-license.php\n\tAuthor Tobias Koppers @sokra\n*/\n\n"use strict";\n\nconst { SyncHook, AsyncSeriesHook } = require("tapable");\nconst { makeWebpackError } = require("../HookWebpackError");\nconst WebpackError = require("../WebpackError");\n\n/**\n * @template T\n * @template R\n */\nclass AsyncQueue {\n\n\t * @param {Object} options options object\n\t * @param {string=} options.name name of the queue\n\t * @param {number} options.parallelism how many items should be processed at once\n\t * @param {function(T, Callback<R>): void} options.processor async function to process items\n\t */\n\tconstructor({ name, parallelism, processor }) {\n\t\tthis._name = name;\n\t\tthis._parallelism = parallelism;\n\t\tthis._processor = processor;\n\t\t/** @type {Map<T, Callback<R>[]>} */\n\t\tthis._callbacks = new Map();\n\t\t/** @type {Set<T>} */\n\t\tthis._queued = new Set();\n\t\t/** @type {Set<T>} */\n\t\tthis._processing = new Set();\n\t\t/** @type {Map<T, [Error, R]>} */\n\t\tthis._results = new Map();\n\t\tthis._activeTasks = 0;\n\t\tthis._willEnsureProcessing = false;\n\t\t/** @type {Callback<R>[] | undefined} */\n\t\tthis.callbacks = undefined;\n\t\tthis.result = undefined;\n\t\t/** @type {WebpackError | undefined} */\n\t\tthis.error = undefined;\n\t}\n}\n\n/**\n * @template T\n * @template K\n * @template R\n */\nclass AsyncQueue {\n\t/**\n\t * @param {Object} options options object\n\t * @param {string=} options.name name of the queue\n\t * @param {number=} options.parallelism how many items should be processed at once\n\t * @param {AsyncQueue<any, any, any>=} options.parent parent queue, which will have priority over this queue and with shared parallelism\n\t * @param {function(T): K=} options.getKey extract key from item\n\t * @param {function(T, Callback<R>): void} options.processor async function to process items\n\t */\n\t\t\t\tcallback(err);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tconst result = this._results.get(item);\n\t\t\tif (result !== undefined) {\n\t\t\t\tprocess.nextTick(() => callback(result[0], result[1]));\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tlet callbacks = this._callbacks.get(item);\n\t\t\tif (callbacks !== undefined) {\n\t\t\t\tcallbacks.push(callback);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tcallbacks = [callback];\n\t\t\tthis._callbacks.set(item, callbacks);\n\t\t\tif (this._stopped) {\n\t\t\t\tthis.hooks.added.call(item);\n\t\t\t\tthis._activeTasks++;\n\t\t\t\tthis._root._children = [this];\n\t\t\t} else {\n\t\t\t\tthis._root._children.push(this);\n\t\t\t}\n\t\t}\n\n\t\tthis.hooks = {\n\t\t\t/** @type {AsyncSeriesHook<[T]>} */\n\t\t\tbeforeAdd: new AsyncSeriesHook(["item"]),\n\t\t\t/** @type {SyncHook<[T]>} */\n\t\t\tadded: new SyncHook(["item"]),\n\t\t\t/** @type {AsyncSeriesHook<[T]>} */\n\t\t\tbeforeStart: new AsyncSeriesHook(["item"]),\n\t\t\t/** @type {SyncHook<[T]>} */\n\t * @returns {void}\n\t */\n\tinvalidate(item) {\n\t\tthis._results.delete(item);\n\t}\n\n\t/**\n\t/**\n\t * @param {T} item an item\n\t * @param {Callback<R>} callback callback function\n\t * @returns {void}\n\t */\n\tadd(item, callback) {\n\t\tif (this._stopped) return callback(new WebpackError("Queue was stopped"));\n\t\tthis.hooks.beforeAdd.callAsync(item, err => {\n\t\t\tif (err) {\n\t\t\t\tcallback(\n\t\t\t\t\tmakeWebpackError(err, `AsyncQueue(${this._name}).hooks.beforeAdd`)\n\t\t\t\t);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tconst key = this._getKey(item);\n\t\t\tconst entry = this._entries.get(key);\n\t\t\tif (entry !== undefined) {\n\t\t\t\tif (entry.state === DONE_STATE) {\n\t\t\t\t\tif (inHandleResult++ > 3) {\n\t\t\t\t\t\tprocess.nextTick(() => callback(entry.error, entry.result));\n\t\t\t\t\t} else {\n\t\t\t\t\t\tcallback(entry.error, entry.result);\n\t\t\t\t\t}\n\t\t\t\t\tinHandleResult--;\n\t\t\t\t} else if (entry.callbacks === undefined) {\n\t\t\t\t\tentry.callbacks = [callback];\n\t\t\t\t} else {\n\t\t\t\t\tentry.callbacks.push(callback);\n\t\t\t\t}\n\t\t\t\treturn;\n\t\t\t}\n\t * @returns {boolean} true, if the item is currently being processed\n\t */\n\tisProcessing(item) {\n\t\treturn this._processing.has(item);\n\t}\n\n\t/**\n\t\t\t\tthis._entries.set(key, newEntry);\n\t\t\t\tthis._queued.enqueue(newEntry);\n\t\t\t\tconst root = this._root;\n\t\t\t\troot._needProcessing = true;\n\t\t\t\tif (root._willEnsureProcessing === false) {\n\t\t\t\t\troot._willEnsureProcessing = true;\n\t\t\t\t\tsetImmediate(root._ensureProcessing);\n\t\t\t\t}\n\t\t\t\tthis.hooks.added.call(item);\n\t * @returns {boolean} true, if the item is currently queued\n\t */\n\tisDone(item) {\n\t\treturn this._results.has(item);\n\t}\n\n\t/**\n\tinvalidate(item) {\n\t\tconst key = this._getKey(item);\n\t\tconst entry = this._entries.get(key);\n\t\tthis._entries.delete(key);\n\t\tif (entry.state === QUEUED_STATE) {\n\t\t\tthis._queued.delete(entry);\n\t\t}\n\t\tfor (const item of this._queued) {\n\t\t\tthis._activeTasks++;\n\t\t\tthis._queued.delete(item);\n\t\t\tthis._processing.add(item);\n\t\t\tthis._startProcessing(item);\n\t\t\tif (this._activeTasks >= this._parallelism) {\n\t\t\t\tthis._willEnsureProcessing = false;\n\twaitFor(item, callback) {\n\t\tconst key = this._getKey(item);\n\t\tconst entry = this._entries.get(key);\n\t\tif (entry === undefined) {\n\t\t\treturn callback(\n\t\t\t\tnew WebpackError(\n\t\t\t\t\t"waitFor can only be called for an already started item"\n\t\t\t\t)\n\t\t\t);\n\t\t}\n\t\tif (entry.state === DONE_STATE) {\n\t\t\tprocess.nextTick(() => callback(entry.error, entry.result));\n\t\t} else if (entry.callbacks === undefined) {\n\t\t\tentry.callbacks = [callback];\n\t\t} else {\n\t\t\tentry.callbacks.push(callback);\n\t\t}\n\t}\n\n\t/**\n\t * @returns {void}\n\t */\n\tstop() {\n\t\tthis._stopped = true;\n\t\tconst queue = this._queued;\n\t\tthis._queued = new ArrayQueue();\n\t\tconst root = this._root;\n\t\tfor (const entry of queue) {\n\t\t\tthis._entries.delete(this._getKey(entry.item));\n\t\t\troot._activeTasks++;\n\t\t\tthis._handleResult(entry, new WebpackError("Queue was stopped"));\n\t\t}\n\t}\n\n\t/**\n\t * @returns {void}\n\t */\n\t\tthis.hooks.result.callAsync(item, err, result, hookError => {\n\t\t\tconst error = hookError || err;\n\n\t\t\tconst callbacks = this._callbacks.get(item);\n\t\t\tthis._processing.delete(item);\n\t\t\tthis._results.set(item, [error, result]);\n\t\t\tthis._callbacks.delete(item);\n\t\t\tthis._activeTasks--;\n\n\t\t\tif (this._willEnsureProcessing === false && this._queued.size > 0) {\n\t * @returns {void}\n\t */\n\tdecreaseParallelism() {\n\t\tconst root = this._root;\n\t\troot._parallelism--;\n\t}\n\n\t/**\n\t * @param {T} item an item\n\t * @returns {boolean} true, if the item is currently being processed\n\t */\n\tisProcessing(item) {\n\t\tconst key = this._getKey(item);\n\t\tconst entry = this._entries.get(key);\n\t\treturn entry !== undefined && entry.state === PROCESSING_STATE;\n\t}\n\n\t/**\n\t * @param {T} item an item\n\t * @returns {boolean} true, if the item is currently queued\n\t */\n\tisQueued(item) {\n\t\tconst key = this._getKey(item);\n\t\tconst entry = this._entries.get(key);\n\t\treturn entry !== undefined && entry.state === QUEUED_STATE;\n\t}\n\n\t/**\n\t * @param {T} item an item\n\t * @returns {boolean} true, if the item is currently queued\n\t */\n\tisDone(item) {\n\t\tconst key = this._getKey(item);\n\t\tconst entry = this._entries.get(key);\n\t\treturn entry !== undefined && entry.state === DONE_STATE;\n\t}\n\n\t/**\n\t * @returns {void}\n\t */\n\t_ensureProcessing() {\n\t\twhile (this._activeTasks < this._parallelism) {\n\t\t\tconst entry = this._queued.dequeue();\n\t\t\tif (entry === undefined) break;\n\t\t\tthis._activeTasks++;\n\t\t\tentry.state = PROCESSING_STATE;\n\t\t\tthis._startProcessing(entry);\n\t\t}\n\t\tthis._willEnsureProcessing = false;\n\t\tif (this._queued.length > 0) return;\n\t\tif (this._children !== undefined) {\n\t\t\tfor (const child of this._children) {\n\t\t\t\twhile (this._activeTasks < this._parallelism) {\n\t\t\t\t\tconst entry = child._queued.dequeue();\n\t\t\t\t\tif (entry === undefined) break;\n\t\t\t\t\tthis._activeTasks++;\n\t\t\t\t\tentry.state = PROCESSING_STATE;\n\t\t\t\t\tchild._startProcessing(entry);\n\t\t\t\t}\n\t\t\t\tif (child._queued.length > 0) return;\n\t\t\t}\n\t\t}\n\t\tif (!this._willEnsureProcessing) this._needProcessing = false;\n\t}\n\n\t/**\n\t * @param {AsyncQueueEntry<T, K, R>} entry the entry\n\t * @returns {void}\n\t */\n\t_startProcessing(entry) {\n\t\tthis.hooks.beforeStart.callAsync(entry.item, err => {\n\t\t\tif (err) {\n\t\t\t\tthis._handleResult(\n\t\t\t\t\tentry,\n\t\t\t\t\tmakeWebpackError(err, `AsyncQueue(${this._name}).hooks.beforeStart`)\n\t\t\t\t);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tlet inCallback = false;\n\t\t\ttry {\n\t\t\t\tthis._processor(entry.item, (e, r) => {\n\t\t\t\t\tinCallback = true;\n\t\t\t\t\tthis._handleResult(entry, e, r);\n\t\t\t\t});\n\t\t\t} catch (err) {\n\t\t\t\tif (inCallback) throw err;\n\t\t\t\tthis._handleResult(entry, err, null);\n\t\t\t}\n\t\t\tthis.hooks.started.call(entry.item);\n\t\t});\n\t}\n\n\t/**\n\t * @param {AsyncQueueEntry<T, K, R>} entry the entry\n\t * @param {WebpackError=} err error, if any\n\t * @param {R=} result result, if any\n\t * @returns {void}\n\t */\n\t_handleResult(entry, err, result) {\n\t\tthis.hooks.result.callAsync(entry.item, err, result, hookError => {\n\t\t\tconst error = hookError\n\t\t\t\t? makeWebpackError(hookError, `AsyncQueue(${this._name}).hooks.result`)\n\t\t\t\t: err;\n\n\t\t\tconst callback = entry.callback;\n\t\t\tconst callbacks = entry.callbacks;\n\t\t\tentry.state = DONE_STATE;\n\t\t\tentry.callback = undefined;\n\t\t\tentry.callbacks = undefined;\n\t\t\tentry.result = result;\n\t\t\tentry.error = error;\n\n\t\t\tconst root = this._root;\n\t\t\troot._activeTasks--;\n\t\t\tif (root._willEnsureProcessing === false && root._needProcessing) {\n\t\t\t\troot._willEnsureProcessing = true;\n\t\t\t\tsetImmediate(root._ensureProcessing);\n\t\t\t}\n\n\t\t\tif (inHandleResult++ > 3) {\n\t\t\t\tprocess.nextTick(() => {\n\t\t\t\t\tcallback(error, result);\n\t\t\t\t\tif (callbacks !== undefined) {\n\t\t\t\t\t\tfor (const callback of callbacks) {\n\t\t\t\t\t\t\tcallback(error, result);\n\t\t\t\t\t\t}\n\t\t\t\t\t}\n\t\t\t\t});\n\t\t\t} else {\n\t\t\t\tcallback(error, result);\n\t\t\t\tif (callbacks !== undefined) {\n\t\t\t\t\tfor (const callback of callbacks) {\n\t\t\t\t\t\tcallback(error, result);\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t}\n\t\t\tinHandleResult--;\n\t\t});\n\t}\n\n\tclear() {\n\t\tthis._entries.clear();\n\t\tthis._queued.clear();\n\t\tthis._activeTasks = 0;\n\t\tthis._willEnsureProcessing = false;\n\t\tthis._needProcessing = false;\n\t\tthis._stopped = false;\n\t}\n}\n\nmodule.exports = AsyncQueue;\n', 'message': 'make Compilation.addModule async and queued\n', 'author_name': 'Tobias Koppers', 'lang': 'js'}