diff --git a/src/engine.ts b/src/engine.ts index cad31744..92673458 100644 --- a/src/engine.ts +++ b/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); diff --git a/test/js/fix/708.test.js b/test/js/fix/708.test.js new file mode 100644 index 00000000..648e8896 --- /dev/null +++ b/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'); + }); +});