Add pooling and request limiting to prevent memory leaks

packages/interactive-code-block/src/lib/components/php-runner/php-loader.ts

@@ -1,8 +1,5 @@
 import type { LoadingStatus } from '../../types';
-import {
-	consumeAPI,
-	spawnPHPWorkerThread,
-} from '@php-wasm/web';
+import { consumeAPI, spawnPHPWorkerThread } from '@php-wasm/web';
 
 import type { PHPClient } from '../../php-worker';
 
@@ -24,9 +21,7 @@ export class PHPLoader extends EventTarget {
 			/** @ts-ignore */
 			'../../php-worker.ts?url&worker'
 		);
-		const worker = await spawnPHPWorkerThread(
-			workerScriptUrl
-		);
+		const worker = await spawnPHPWorkerThread(workerScriptUrl);
 		const php = consumeAPI<PHPClient>(worker);
 		php?.onDownloadProgress((e) => {
 			const { loaded, total } = e.detail;

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/public/with-node-version.js

@@ -6,9 +6,7 @@ import fs from 'fs';
 // Set the minimum required/supported version of node here.
 
 // Check if `--blueprint=` is passed in proccess.argv
-const hasBlueprint = process.argv.some((arg) =>
-	arg.startsWith('--blueprint=')
-);
+const hasBlueprint = process.argv.some((arg) => arg.startsWith('--blueprint='));
 
 const minimum = {
 	// `--blueprint=` requires node v20

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,261 @@
+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 {
+	instanceInfo = 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,
+	} = {}) {
+		this.spawner = spawner;
+		this.maxRequests = maxRequests;
+		this.maxJobs = maxJobs;
+		this[Reap]();
+		this[Spawn]();
+	}
+
+	/**
+	 * Find the next available idle instance.
+	 */
+	getIdleInstance() {
+		const sorted = [...this.instanceInfo].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 the notifier so async calling code
+			// can still respond correctly when the item
+			// is finally processed.
+			return notifier;
+		} else {
+			// If we've got an instance available, run the provided callback.
+
+			// Given an instance, create a new callback that will clean up
+			// after the instance processes a request, and optionally
+			// will also kick off the next request.
+			const onCompleted = (instance) => async () => {
+				this.running.delete(instance);
+
+				this[Reap]();
+				const newInstances = this[Spawn]();
+
+				// Break out here if the backlog is empty.
+				if (!this.backlog.length) {
+					return;
+				}
+
+				// This is the instance that completed
+				// so we can re-use it...
+				let nextInstance = instance;
+
+				// ... but, if we've just spanwed a fresh
+				// instance, use that one instead.
+				if (newInstances.size)
+					for (const instance of newInstances) {
+						nextInstance = instance;
+						break;
+					}
+
+				const next = this.backlog.shift();
+				const info = this.instanceInfo.get(nextInstance);
+
+				this.running.add(nextInstance);
+				info.requests++;
+
+				let request;
+
+				try {
+					// Don't ACTUALLY do anything until the
+					// instance is done spawning.
+					request = next(await nextInstance);
+				}
+				catch(error) {
+					// Re-queue the request if the instance
+					// failed initialization.
+					this.backlog.unshift(next);
+
+					// Catch any errors and log to the console.
+					// Deactivate the instance.
+					this[Fatal](nextInstance, error);
+
+					// Return the notifier so async calling code
+					// can still respond correctly when the item
+					// is finally processed.
+					return this.notifiers.get(next);
+				}
+
+				const completed = onCompleted(nextInstance);
+
+				// Make sure onComplete & running.delete run
+				// no matter how the request resolves.
+				request.finally(() => {
+					this.running.delete(nextInstance);
+					completed();
+				});
+
+				// Grab the accept handler from the notfier
+				// promise and run it if the request resolves.
+				request.then((ret) => {
+					const notifier = this.notifiers.get(next);
+					this.notifiers.delete(next);
+					notifier.accept(ret);
+				});
+
+				// Grab the reject handler from the notfier
+				// promise and run it if the request rejects.
+				request.catch((error) => {
+					const notifier = this.notifiers.get(next);
+					this.notifiers.delete(next);
+					notifier.reject(error);
+					// Catch any errors and log to the console.
+					// Deactivate the instance.
+					this[Fatal](nextInstance, error);
+				});
+			};
+
+			const info = this.instanceInfo.get(idleInstance);
+
+			this.running.add(idleInstance);
+			info.requests++;
+
+			let request;
+
+			try {
+				// Don't ACTUALLY do anything until the
+				// instance is done spawning.
+				request = item(await idleInstance);
+			}
+			catch(error) {
+				// Re-queue the request if the instance
+				// failed initialization.
+				this.backlog.unshift(item);
+
+				// Catch any errors and log to the console.
+				// Deactivate the instance.
+				this[Fatal](idleInstance, error);
+
+				// 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 the notifier so async calling code
+				// can still respond correctly when the item
+				// is finally processed.
+				return notifier;
+			}
+
+			// Make sure onComplete runs no matter how the request resolves.
+			request.finally(onCompleted(idleInstance));
+
+			// Catch any errors and log to the console.
+			// Deactivate the instance.
+			request.catch((error) => this[Fatal](idleInstance, error));
+
+			return request;
+		}
+	}
+
+	/**
+	 * PRIVATE
+	 * Spawns new instances if the pool is not full.
+	 * Returns a list of new instances.
+	 */
+	[Spawn]() {
+		const newInstances = new Set();
+		while (this.maxJobs > 0 && this.instanceInfo.size < this.maxJobs) {
+			const info = new PoolInfo();
+			const instance = this.spawner();
+			this.instanceInfo.set(instance, info);
+			info.active = true;
+			newInstances.add(instance);
+		}
+		return newInstances;
+	}
+
+	/**
+	 * PRIVATE
+	 * Reaps children if they've passed the maxRequest count.
+	 */
+	[Reap]() {
+		for (const [instance, info] of this.instanceInfo) {
+			if (this.maxRequests > 0 && info.requests >= this.maxRequests) {
+				info.active = false;
+				this.instanceInfo.delete(instance);
+				// instance.then(unwrapped => unwrapped.destroy());
+				continue;
+			}
+		}
+	}
+
+	/**
+	 * PRIVATE
+	 * Handle fatal errors gracefully.
+	 */
+	[Fatal](instance, error) {
+		console.error(error);
+		if (this.instanceInfo.has(instance)) {
+			const info = this.instanceInfo.get(instance);
+			info.active = false;
+			this.instanceInfo.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);