Implement flow-control for test worker IPC

Author: novemberborn

lib/cli.js

@@ -430,14 +430,17 @@ exports.run = async () => { // eslint-disable-line complexity
 		reporter.startRun(plan);
 
 		if (process.env.AVA_EMIT_RUN_STATUS_OVER_IPC === 'I\'ll find a payphone baby / Take some time to talk to you') {
+			const {controlFlow} = require('./ipc-flow-control');
+			const bufferedSend = controlFlow(process);
+
 			if (process.versions.node >= '12.16.0') {
 				plan.status.on('stateChange', evt => {
-					process.send(evt);
+					bufferedSend(evt);
 				});
 			} else {
 				const v8 = require('v8');
 				plan.status.on('stateChange', evt => {
-					process.send([...v8.serialize(evt)]);
+					bufferedSend([...v8.serialize(evt)]);
 				});
 			}
 		}

lib/fork.js

@@ -3,6 +3,7 @@ const childProcess = require('child_process');
 const path = require('path');
 const fs = require('fs');
 const Emittery = require('emittery');
+const {controlFlow} = require('./ipc-flow-control');
 
 if (fs.realpathSync(__filename) !== __filename) {
 	console.warn('WARNING: `npm link ava` and the `--preserve-symlink` flag are incompatible. We have detected that AVA is linked via `npm link`, and that you are using either an early version of Node 6, or the `--preserve-symlink` flag. This breaks AVA. You should upgrade to Node 6.2.0+, avoid the `--preserve-symlink` flag, or avoid using `npm link ava`.');
@@ -45,12 +46,12 @@ module.exports = (file, options, execArgv = process.execArgv) => {
 		emitStateChange({type: 'worker-stderr', chunk});
 	});
 
+	const bufferedSend = controlFlow(subprocess);
+
 	let forcedExit = false;
 	const send = evt => {
-		if (subprocess.connected && !finished && !forcedExit) {
-			subprocess.send({ava: evt}, () => {
-				// Disregard errors.
-			});
+		if (!finished && !forcedExit) {
+			bufferedSend({ava: evt});
 		}
 	};
 

lib/ipc-flow-control.js

@@ -0,0 +1,39 @@
+function controlFlow(channel) {
+	let errored = false;
+	let deliverImmediately = true;
+
+	const backlog = [];
+	const deliverNext = error => {
+		if (error !== null) {
+			errored = true;
+		}
+
+		if (errored || !channel.connected) {
+			backlog.length = 0; // Free memory.
+			return; // We can't send.
+		}
+
+		let ok = true;
+		while (ok && backlog.length > 0) { // Stop sending after backpressure.
+			ok = channel.send(backlog.shift(), deliverNext);
+		}
+
+		// Re-enable immediate delivery if there is no backpressure and the backlog
+		// has been cleared.
+		deliverImmediately = ok && backlog.length === 0;
+	};
+
+	return message => {
+		if (errored || !channel.connected) {
+			return;
+		}
+
+		if (deliverImmediately) {
+			deliverImmediately = channel.send(message, deliverNext);
+		} else {
+			backlog.push(message);
+		}
+	};
+}
+
+exports.controlFlow = controlFlow;

lib/worker/ipc.js

@@ -1,5 +1,6 @@
 'use strict';
 const Emittery = require('emittery');
+const {controlFlow} = require('../ipc-flow-control');
 
 const emitter = new Emittery();
 process.on('message', message => {
@@ -25,10 +26,9 @@ process.on('message', message => {
 exports.options = emitter.once('options');
 exports.peerFailed = emitter.once('peerFailed');
 
+const bufferedSend = controlFlow(process);
 function send(evt) {
-	if (process.connected) {
-		process.send({ava: evt});
-	}
+	bufferedSend({ava: evt});
 }
 
 exports.send = send;