From 5aefdd1b8582f8e9c7c90bf4eb321cf1f13841a4 Mon Sep 17 00:00:00 2001 From: Mathias Wulff Date: Thu, 22 Oct 2026 03:11:39 +1100 Subject: [PATCH 1/3] Limit how many files are handled at the same time to close #708 --- README.md | 1 + src/engine.ts | 37 ++++++++++++------- test/js/fix/708.test.js | 78 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 13 deletions(-) create mode 100644 test/js/fix/708.test.js diff --git a/README.md b/README.md index b18f9400..8a156cf5 100644 --- a/README.md +++ b/README.md @@ -288,6 +288,7 @@ Use --js-pipe to access the pipe data in your javascript code. ### Limitations - RexReplace reads each file fully into memory, so working on your 8Gb log files will probably not be ideal. +- At most 16 files (or two per CPU core, if that is more) are handled at the same time, so many files do not fill up the memory or hit the limit of open files. - For versions of Node prior to 6, please use version 2.2.x. For versions of Node prior to 0.12, please use [the legacy version of RexReplace called rreplace](https://www.npmjs.com/package/rreplace) ### Priorities diff --git a/src/engine.ts b/src/engine.ts index cad31744..31f4c98c 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,39 @@ 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. + const workers = config.voidAsync ? 1 : 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..cb740c31 --- /dev/null +++ b/test/js/fix/708.test.js @@ -0,0 +1,78 @@ +// 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('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'); + }); +}); From 3e72ea813d3e21650aa72a23c3793e2b2b2e43c9 Mon Sep 17 00:00:00 2001 From: Mathias Wulff Date: Thu, 22 Oct 2026 03:18:31 +1100 Subject: [PATCH 2/3] Remove the README line about the number of files handled at the same time --- README.md | 1 - 1 file changed, 1 deletion(-) diff --git a/README.md b/README.md index 8a156cf5..b18f9400 100644 --- a/README.md +++ b/README.md @@ -288,7 +288,6 @@ Use --js-pipe to access the pipe data in your javascript code. ### Limitations - RexReplace reads each file fully into memory, so working on your 8Gb log files will probably not be ideal. -- At most 16 files (or two per CPU core, if that is more) are handled at the same time, so many files do not fill up the memory or hit the limit of open files. - For versions of Node prior to 6, please use version 2.2.x. For versions of Node prior to 0.12, please use [the legacy version of RexReplace called rreplace](https://www.npmjs.com/package/rreplace) ### Priorities From cd4f4d47f6aa08413a9b857af5d7365f6813c0bc Mon Sep 17 00:00:00 2001 From: Mathias Wulff Date: Thu, 22 Oct 2026 03:26:25 +1100 Subject: [PATCH 3/3] Only ask for the number of CPU cores when there are more than 16 files --- src/engine.ts | 8 ++++++-- test/js/fix/708.test.js | 11 +++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/src/engine.ts b/src/engine.ts index 31f4c98c..92673458 100644 --- a/src/engine.ts +++ b/src/engine.ts @@ -100,8 +100,12 @@ export async function engine(config: any = {engine: 'default'}) { ); // 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. - const workers = config.voidAsync ? 1 : Math.max(16, os.availableParallelism() * 2); + // 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) { diff --git a/test/js/fix/708.test.js b/test/js/fix/708.test.js index cb740c31..648e8896 100644 --- a/test/js/fix/708.test.js +++ b/test/js/fix/708.test.js @@ -49,6 +49,17 @@ describe('#708 the number of files in work at the same time is limited', () => { 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', {