Limit how many files are handled at the same time to close #708

src/engine.ts

@@ -1,5 +1,7 @@
 const fs = require('fs');
 
+const os = require('os');
+
 const path = require('path');
 
 const globs = require('fast-glob');
@@ -87,30 +89,43 @@ export async function engine(config: any = {engine: 'default'}) {
 
 	chat(config.files.length + ' files found');
 
-	// Ensure predictable sequence of output
-	if (config.files.length > 1) {
-		config.voidAsync = config.voidAsync || config.outputMatch || config.output;
-	}
+	// One file needs no async flow, and with -o or -m one file at a time gives the output in order
+	config.voidAsync =
+		config.voidAsync || 1 === config.files.length || config.output || config.outputMatch;
 
 	step(config);
 
-	const work = config.files
-		// Correct filepath
-		//.map(filepath=>path.normalize(process.cwd()+'/'+filepath))
-		// Find out if any filepaths are invalid
-		.filter((filepath) => (fs.existsSync(filepath) ? true : error('File not found:', filepath)))
+	const files = config.files.filter((filepath) =>
+		fs.existsSync(filepath) ? true : error('File not found:', filepath)
+	);
 
-		// Do the replacement
-		.map((filepath) => openFile(filepath, config));
+	// A limited number of files are in work at the same time, so the memory and the open files stay within bounds.
+	// Each worker takes the next file when it is done with one. Up to 16 files are all handled at once.
+	const workers = config.voidAsync
+		? 1
+		: files.length <= 16
+			? files.length
+			: Math.max(16, os.availableParallelism() * 2);
+	let next = 0;
+	const worker = async () => {
+		while (next < files.length) {
+			await openFile(files[next++], config);
+		}
+	};
 
-	await Promise.all(work);
+	await Promise.all(Array.from({length: workers}, worker));
 	return printDone();
 }
 
 function openFile(file, config) {
 	if (config.voidAsync) {
 		chat('Open sync: ' + file);
-		var data = fs.readFileSync(file, config.encoding);
+		let data;
+		try {
+			data = fs.readFileSync(file, config.encoding);
+		} catch (e) {
+			return error(e);
+		}
 		return doReplacement(file, config, data);
 	} else {
 		chat('Open async: ' + file);

test/js/fix/708.test.js

@@ -0,0 +1,89 @@
+// https://github.com/mathiasrw/rexreplace/issues/708
+// All files were in work at the same time, so with many files the memory filled up and files failed with `EMFILE: too many open files`.
+import {describe, test, expect, spyOn} from 'bun:test';
+import fs from 'fs';
+import os from 'os';
+import {rr} from '../helper.js';
+
+const workers = Math.max(16, os.availableParallelism() * 2);
+
+// Make `count` files with the content `foo` and give their names
+async function makeFiles(count) {
+	const names = Array.from({length: count}, (_, i) => `file-${i}.txt`);
+	await Promise.all(names.map((name) => Bun.write(name, 'foo')));
+	return names;
+}
+
+describe('#708 the number of files in work at the same time is limited', () => {
+	test('all files are changed when there are more files than workers', async () => {
+		const names = await makeFiles(workers * 3);
+		const result = await rr('o', 'x', {includeGlob: ['*.txt'], voidAsync: false});
+		expect(result.err).toBe('');
+		expect(result.out.split('\n').sort()).toEqual(['', ...names].sort());
+		for (const name of names) {
+			expect(await Bun.file(name).text()).toBe('fxx');
+		}
+	});
+
+	test(`at most ${workers} files are read at the same time`, async () => {
+		await makeFiles(workers * 3);
+		const readFile = fs.promises.readFile;
+		let reading = 0;
+		let mostReading = 0;
+		const spy = spyOn(fs.promises, 'readFile').mockImplementation(async (...args) => {
+			mostReading = Math.max(mostReading, ++reading);
+			try {
+				return await readFile(...args);
+			} finally {
+				reading--;
+			}
+		});
+
+		try {
+			await rr('o', 'x', {includeGlob: ['*.txt'], voidAsync: false});
+		} finally {
+			spy.mockRestore();
+		}
+
+		expect(mostReading).toBeGreaterThan(1);
+		expect(mostReading).toBeLessThanOrEqual(workers);
+	});
+
+	test('the number of CPU cores is not asked for with 16 files or fewer', async () => {
+		await makeFiles(16);
+		const spy = spyOn(os, 'availableParallelism');
+		try {
+			await rr('o', 'x', {includeGlob: ['*.txt'], voidAsync: false});
+			expect(spy).not.toHaveBeenCalled();
+		} finally {
+			spy.mockRestore();
+		}
+	});
+
+	test('one file is handled in the synchronous flow', async () => {
+		await Bun.write('my.file', 'foo');
+		const result = await rr('o', 'x', {
+			includeGlob: ['my.file'],
+			voidAsync: false,
+			verbose: true,
+		});
+		expect(result.err).toContain('Open sync: my.file');
+		expect(await Bun.file('my.file').text()).toBe('fxx');
+	});
+
+	test('a file that can not be read gives an error in the synchronous flow, and the run goes on', async () => {
+		await Bun.write('locked.file', 'foo');
+		await Bun.write('my.file', 'foo');
+		fs.chmodSync('locked.file', 0);
+
+		const result = await rr('o', 'x', {
+			includeGlob: ['locked.file', 'my.file'],
+			voidAsync: true,
+		});
+
+		expect(result.out).toBe('my.file\n');
+		expect(result.err).toContain('EACCES');
+		expect(result.exit).toBe(null);
+		expect(await Bun.file('my.file').text()).toBe('fxx');
+	});
+});