Add pooling and request limiting to prevent memory leaks

packages/nx-extensions/src/executors/built-script/executor.ts

@@ -7,6 +7,7 @@ const dirname = __dirname;
 
 export default async function runExecutor(options: BuiltScriptExecutorSchema) {
 	const args = [
+		...(options.debug ? ['--inspect-brk'] : []),
 		'--loader',
 		join(dirname, 'loader.mjs'),
 		options.scriptPath,

packages/nx-extensions/src/executors/built-script/schema.d.ts

@@ -1,4 +1,5 @@
 export interface BuiltScriptExecutorSchema {
 	scriptPath: string;
+	debug: boolean;
 	__unparsed__: string;
 } // eslint-disable-line

packages/nx-extensions/src/executors/built-script/schema.json

@@ -10,6 +10,11 @@
 			"description": "Path of the script to run.",
 			"x-prompt": "What script would you like to run?"
 		},
+		"debug": {
+			"type": "boolean",
+			"description": "Use devtools as a debugger.",
+			"x-prompt": "Would you like to use devtools?"
+		},
 		"__unparsed__": {
 			"hidden": true,
 			"type": "array",

packages/wp-now/src/config.ts

@@ -20,6 +20,7 @@ export interface CliOptions {
 	port?: number;
 	blueprint?: string;
 	reset?: boolean;
+	maxRequests?: number;
 }
 
 export const enum WPNowMode {
@@ -43,6 +44,7 @@ export interface WPNowOptions {
 	wpContentPath?: string;
 	wordPressVersion?: string;
 	numberOfPhpInstances?: number;
+	maxRequests?: number;
 	blueprintObject?: Blueprint;
 	reset?: boolean;
 }
@@ -54,6 +56,7 @@ export const DEFAULT_OPTIONS: WPNowOptions = {
 	projectPath: process.cwd(),
 	mode: WPNowMode.AUTO,
 	numberOfPhpInstances: 1,
+	maxRequests: 512,
 	reset: false,
 };
 
@@ -111,6 +114,7 @@ export default async function getWpNowConfig(
 		phpVersion: args.php as SupportedPHPVersion,
 		projectPath: args.path as string,
 		wordPressVersion: args.wp as string,
+		maxRequests: args.maxRequests as number,
 		port,
 		reset: args.reset as boolean,
 	};

packages/wp-now/src/pool.ts

@@ -0,0 +1,180 @@
+import { NodePHP } from '@php-wasm/node';
+
+const Fatal = Symbol('Fatal');
+const Spawn = Symbol('Spawn');
+const Reap = Symbol('Reap');
+
+let childCount = 0;
+
+export class PoolInfo {
+	id = childCount++;
+	requests = 0;
+	started = Date.now();
+	active = false;
+}
+
+/**
+ * Maintains and refreshes a list of php instances
+ * such that each one will only be fed X number of requests
+ * before being discarded and replaced.
+ */
+export class Pool {
+	instances = new Map(); // php => PoolInfo
+
+	spawner: () => Promise<any>; // Callback to create new instances.
+	maxRequests: number; // Max requests to feed each instance
+	maxJobs: number; // Max number of instances to maintain at once.
+
+	notifiers = new Map(); // Inverted promises to notify async code of backlogged item processed.
+	running = new Set(); // Set of busy PHP instances.
+	backlog = []; // Set of request callbacks waiting to be run.
+
+	constructor({
+		spawner = async (): Promise<any> => {},
+		maxRequests = 2000,
+		maxJobs = 5,
+	} = {}) {
+		Object.assign(this, { spawner, maxRequests, maxJobs });
+		this[Reap]();
+		this[Spawn]();
+	}
+
+	/**
+	 * Find the next available idle instance.
+	 */
+	getIdleInstance() {
+		const sorted = [...this.instances].sort(
+			(a, b) => a[1].requests - b[1].requests
+		);
+
+		for (const [instance, info] of sorted) {
+			if (this.running.has(instance)) {
+				continue;
+			}
+
+			if (!info.active) {
+				continue;
+			}
+			return instance;
+		}
+
+		return false;
+	}
+
+	/**
+	 * Queue up a callback that will make a request when an
+	 * instance becomes idle.
+	 */
+	async enqueue(item: (php: NodePHP) => Promise<any>) {
+		const idleInstance = this.getIdleInstance();
+
+		if (!idleInstance) {
+			// Defer the callback if we don't have an idle instance available.
+			this.backlog.push(item);
+
+			// Split a promise open so it can be accepted or
+			// rejected later when the item is processed.
+			const notifier = new Promise((accept, reject) =>
+				this.notifiers.set(item, [accept, reject])
+			);
+
+			return notifier;
+		} else {
+			// If we've got an instance available, run the provided callback.
+
+			// When the provided callback completes, check to see if
+			// any more requests have been added to the pool
+			const onCompleted = async () => {
+				this.running.delete(idleInstance);
+
+				this[Reap]();
+				this[Spawn]();
+
+				if (!this.backlog.length) {
+					return;
+				}
+
+				const idleInstanceNext = this.getIdleInstance();
+
+				const next = this.backlog.shift();
+				const info = this.instances.get(idleInstanceNext);
+
+				info.requests++;
+
+				const request = next(await idleInstanceNext);
+
+				request.finally(onCompleted);
+
+				request.then((ret) => {
+					const notifier = this.notifiers.get(next);
+					this.notifiers.delete(next);
+					notifier[0](ret);
+				});
+
+				request.catch((error) => {
+					const notifier = this.notifiers.get(next);
+					this.notifiers.delete(next);
+					notifier[1](error);
+					this[Fatal](idleInstanceNext, error);
+				});
+
+				this.running.add(idleInstance);
+			};
+
+			const info = this.instances.get(idleInstance);
+
+			info.requests++;
+
+			this.running.add(idleInstance);
+
+			const request = item(await idleInstance);
+
+			request.catch((error) => this[Fatal](idleInstance, error));
+
+			// Make sure onComplete runs no matter how the request resolves
+			request.finally(onCompleted);
+
+			return request;
+		}
+	}
+
+	/**
+	 * PRIVATE
+	 * Spawns new instances if the pool is not full.
+	 */
+	[Spawn]() {
+		while (this.maxJobs > 0 && this.instances.size < this.maxJobs) {
+			const info = new PoolInfo();
+			const instance = this.spawner();
+			this.instances.set(instance, info);
+			info.active = true;
+		}
+	}
+
+	/**
+	 * PRIVATE
+	 * Reaps children if they've passed the maxRequest count.
+	 */
+	[Reap]() {
+		for (const [instance, info] of this.instances) {
+			if (this.maxRequests > 0 && info.requests >= this.maxRequests) {
+				info.active = false;
+				this.instances.delete(instance);
+				continue;
+			}
+		}
+	}
+
+	/**
+	 * PRIVATE
+	 * Handle fatal errors gracefully.
+	 */
+	[Fatal](instance, error) {
+		console.error(error);
+		if (this.instances.has(instance)) {
+			const info = this.instances.get(instance);
+			info.active = false;
+			this.instances.delete(instance);
+		}
+	}
+}

packages/wp-now/src/run-cli.ts

@@ -76,6 +76,11 @@ export async function runCli() {
 						'Create a new project environment, destroying the old project environment.',
 					type: 'boolean',
 				});
+				yargs.option('maxRequests', {
+					describe:
+						'Max number of requests before refreshing PHP instance.',
+					type: 'number',
+				});
 			},
 			async (argv) => {
 				const spinner = startSpinner('Starting the server...');
@@ -87,6 +92,7 @@ export async function runCli() {
 						port: argv.port as number,
 						blueprint: argv.blueprint as string,
 						reset: argv.reset as boolean,
+						maxRequests: argv.maxRequests as number,
 					});
 					portFinder.setPort(options.port as number);
 					const { url } = await startServer(options);

packages/wp-now/src/start-server.ts

@@ -62,7 +62,7 @@ export async function startServer(
 	app.use(compression({ filter: shouldCompress }));
 	app.use(addTrailingSlash('/wp-admin'));
 	const port = await portFinder.getOpenPort();
-	const { php, options: wpNowOptions } = await startWPNow(options);
+	const { php, options: wpNowOptions, pool } = await startWPNow(options);
 
 	app.use('/', async (req, res) => {
 		try {
@@ -103,11 +103,15 @@ export async function startServer(
 				),
 				body: body as string,
 			};
-			const resp = await php.request(data);
+
+			const resp = await pool.enqueue((php) => php.request(data));
+
 			res.statusCode = resp.httpStatusCode;
-			Object.keys(resp.headers).forEach((key) => {
-				res.setHeader(key, resp.headers[key]);
-			});
+
+			Object.keys(resp.headers).forEach((key) =>
+				res.setHeader(key, resp.headers[key])
+			);
+
 			res.end(resp.bytes);
 		} catch (e) {
 			output?.trace(e);