process-manager.js

/**
 * Process Manager
 * Pokemon Showdown - http://pokemonshowdown.com/
 *
 * This file abstract out multiprocess logic involved in several tasks.
 *
 * @license MIT license
 */

'use strict';

const EventEmitter = require('events');

const processManagers = new Map();

function serialize(str) {
	if (typeof str === 'string') return str;
	return JSON.stringify(str);
}

class ProcessWrapper extends EventEmitter {
	constructor(PM) {
		super();

		this.PM = PM;
		this.active = true;
		this.pendingTasks = new Map();
		this.process = require('child_process').fork(PM.execFile, {cwd: __dirname});

		// Allow events to bubble-up to the wrapper
		this.process.on('message', message => this.emit('message', message));

		this.on('message', PM.onMessageUpstream);
	}

	send(data) {
		return this.process.send(data);
	}

	release() {
		if (this.load || this.active) return;
		this.process.disconnect();
	}

	get load() {
		return this.pendingTasks.size;
	}
}

// execFile - the path to the file to spawn the child process(es) from
// maxProcesses - the maximum number of child processes to spawn
// isChatBased - the process manager handles some chat functionality
class ProcessManager {
	constructor(options) {
		if (!('execFile' in options)) {
			Monitor.debug(
				"No execFile property was missing form the options object to be " +
				"given to the ProcessManager constructor!"
			);
		} else if (!('maxProcesses' in options) || !('isChatBased' in options)) {
			Monitor.debug(
				"An options object given to the ProcessManager constructor is " +
				"missing required properties! The filename given is: " + (options.execFile || '""') + "."
			);
		}

		this.processes = [];
		this.taskId = 0;
		this.execFile = options.execFile;
		this.maxProcesses = (typeof options.maxProcesses === 'number') ? options.maxProcesses : 1;
		this.isChatBased = !!options.isChatBased;

		processManagers.set(this, options.execFile);
	}

	spawn() {
		for (let i = this.processes.length; i < this.maxProcesses; ++i) {
			this.processes.push(new ProcessWrapper(this));
		}
	}

	unspawn() {
		for (let process of this.processes.splice(0)) {
			process.active = false;
			process.release();
		}
	}

	respawn() {
		this.unspawn();
		this.spawn();
	}

	acquire() {
		let process = this.processes[0];
		for (let i = 1; i < this.processes.length; ++i) {
			if (this.processes[i].load < process.load) {
				process = this.processes[i];
			}
		}
		return process;
	}

	release(process) {
		process.release();
	}

	send(...args) {
		if (!this.processes.length) {
			return Promise.resolve(this.receive.apply(this, args));
		}

		let serializedArgs = args.map(serialize).join('|');
		return new Promise((resolve, reject) => {
			let process = this.acquire();
			process.pendingTasks.set(this.taskId, resolve);
			try {
				process.process.send(`${this.taskId}|${serializedArgs}`);
			} catch (e) {}
			++this.taskId;
		});
	}

	sendSync(...args) {
		// synchronously!
		return this.receive.apply(this, args);
	}

	onMessageUpstream() {
		// Expected to resolve the pending task completed.
	}

	onMessageDownstream() {
		// Expected to call `receive()` at some point,
		// and send the result to the parent process.
	}

	receive() {
		// This is where the child process actually does stuff
		// To be overriden by specific implementations.
		return '';
	}
}

ProcessManager.cache = processManagers;
ProcessManager.ProcessWrapper = ProcessWrapper;
module.exports = ProcessManager;