fix: memory leak during search #678
Massimo Melina committed
Jul 17, 2024 at 23:38 UTC
f9b9958fbb4c8df9a0032bf9cb4377fc84c99d4c
5 files changed
+159
-21
src/api.vfs.ts
+3
-1
@@ -199,9 +199,11 @@ const apis: ApiHandlers = {
199
try {
200
const matching = makeMatcher(fileMask)
201
path = isWindowsDrive(path) ? path + '\\' : resolve(path || '/')
202
- for await (const [name, isDir] of dirStream(path)) {
202
+ for await (const entry of dirStream(path)) {
203
if (ctx.req.aborted)
204
return
205
+ const {path:name} = entry
206
+ const isDir = entry.isDirectory()
207
if (!isDir)
208
if (!files || fileMask && !matching(name))
209
continue
src/dirStream.ts
new
+116
@@ -0,0 +1,116 @@
1
+import { makeQ } from './makeQ'
2
+import { opendir, stat } from 'fs/promises'
3
+import { join } from 'path'
4
+import { Readable } from 'stream'
5
+import { pendingPromise } from './cross'
6
+import { Stats, Dirent } from 'node:fs'
7
+
8
+export interface DirStreamEntry extends Dirent {
9
+ closingBranch?: Promise<string>
10
+ stats?: Stats
11
+}
12
+
13
+const dirQ = makeQ(3)
14
+
15
+export function createDirStream(startPath: string, depth=0) {
16
+ let stopped = false
17
+ let started = false
18
+ const closingQ: string[] = []
19
+ const stream = new Readable({
20
+ objectMode: true,
21
+ read() {
22
+ if (started) return
23
+ started = true
24
+ dirQ.add(() => readDir('', depth)
25
+ .then(res => { // don't make the job await for it, but use it to close the stream
26
+ Promise.resolve(res?.branchDone).then(() => {
27
+ stream.push(null)
28
+ })
29
+ }, e => {
30
+ stream.emit('error', e)
31
+ return stream.push(null)
32
+ })
33
+ )
34
+ }
35
+ })
36
+ stream.on('close', () => stopped = true)
37
+ return Object.assign(stream, {
38
+ stop() {
39
+ stopped = true
40
+ if (!dirQ.isWorking())
41
+ stream.push(null)
42
+ },
43
+ })
44
+
45
+ async function readDir(path: string, depth: number) {
46
+ if (stopped) return
47
+ const base = join(startPath, path)
48
+ const dir = await opendir(base)
49
+ const subDirsDone: Promise<any>[] = []
50
+ let last: DirStreamEntry | undefined = undefined
51
+ let n = 0
52
+ for await (let entry of dir) {
53
+ if (stopped) {
54
+ await dir.close().catch(() => {}) // only necessary for early exit
55
+ break
56
+ }
57
+ const stats = entry.isSymbolicLink() && await stat(join(base, entry.name)).catch(() => null)
58
+ if (stats === null) continue
59
+ if (stats)
60
+ entry = new DirentFromStats(entry.name, stats)
61
+ entry.path = (path && path + '/') + entry.name
62
+ if (last && closingQ.length) // pending entries
63
+ last.closingBranch = Promise.resolve(closingQ.shift()!)
64
+ last = entry
65
+ const expanded: DirStreamEntry = entry
66
+ if (stats)
67
+ expanded.stats = stats
68
+ if (depth > 0 && entry.isDirectory()) {
69
+ const branchDone = pendingPromise() // per-job
70
+ const job = () =>
71
+ readDir(entry.path, depth - 1) // recur
72
+ .then(x => x, () => {}) // mute errors
73
+ .then(res => { // don't await, as readDir must resolve without branch being done
74
+ if (!res?.n)
75
+ closingQ.push(entry.path) // no children to tell i'm done
76
+ Promise.resolve(res?.branchDone).then(() =>
77
+ branchDone.resolve())
78
+ })
79
+ dirQ.add(job) // this won't start until next tick
80
+ subDirsDone.push(branchDone)
81
+ }
82
+ stream.push(entry)
83
+ n++
84
+ }
85
+ const branchDone = Promise.allSettled(subDirsDone).then(() => {})
86
+ if (last) // using streams, we don't know when the entries are received, so we need to notify on last item
87
+ last.closingBranch = branchDone.then(() => path)
88
+ else
89
+ closingQ.push(path) // ok, we'll ask next one to carry this info
90
+ // don't return the promise directly, as this job ends here, but communicate to caller the promise for the whole branch
91
+ return { branchDone, n }
92
+ }
93
+}
94
+
95
+type DirentStatsKeysIntersection = keyof Dirent & keyof Stats;
96
+const kStats = Symbol('stats')
97
+// Adapting an internal class in Node.js to mimic the behavior of `Dirent` when creating it manually from `Stats`.
98
+// https://github.com/nodejs/node/blob/a4cf6b204f0b160480153dc293ae748bf15225f9/lib/internal/fs/utils.js#L199C1-L213
99
+export class DirentFromStats extends Dirent {
100
+ private readonly [kStats]: Stats;
101
+ constructor(name: string, stats: Stats) {
102
+ // @ts-expect-error The constructor has parameters, but they are not represented in types.
103
+ // https://github.com/nodejs/node/blob/a4cf6b204f0b160480153dc293ae748bf15225f9/lib/internal/fs/utils.js#L164
104
+ super(name, null);
105
+ this[kStats] = stats;
106
+ }
107
+}
108
+
109
+for (const key of Reflect.ownKeys(Dirent.prototype)) {
110
+ const name = key as DirentStatsKeysIntersection | 'constructor';
111
+ if (name === 'constructor')
112
+ continue;
113
+ DirentFromStats.prototype[name] = function () {
114
+ return this[kStats][name]();
115
+ };
116
+}
\ No newline at end of file
src/makeQ.ts
new
+25
@@ -0,0 +1,25 @@
1
+export function makeQ(parallelization=1) {
2
+ const running = new Set<Promise<unknown>>()
3
+ const queued: Array<() => Promise<unknown>> = []
4
+ return {
5
+ add(toAdd: typeof queued[0]) {
6
+ queued.push(toAdd)
7
+ setTimeout(startNextIfPossible) // avoid nesting/stacking of jobs
8
+ },
9
+ isWorking() { return running.size > 0 },
10
+ isFree() { return running.size < parallelization },
11
+ }
12
+ function startNextIfPossible() {
13
+ while (running.size < parallelization) {
14
+ const job = queued.pop()
15
+ if (!job) break // finished
16
+ const working = job() // start the job
17
+ if (!working) continue // it was canceled
18
+ running.add(working)
19
+ working.then(() => {
20
+ running.delete(working)
21
+ startNextIfPossible()
22
+ })
23
+ }
24
+ }
25
+}
\ No newline at end of file
src/util-files.ts
+7
-17
@@ -8,6 +8,7 @@ import glob from 'fast-glob'
8
import { IS_WINDOWS } from './const'
9
import { runCmd } from './util-os'
10
import { once, Readable } from 'stream'
11
+import { createDirStream, DirStreamEntry } from './dirStream'
12
// @ts-ignore
13
import unzipper from 'unzip-stream'
14
@@ -71,27 +72,16 @@ export function adjustStaticPathForGlob(path: string) {
72
return glob.escapePath(path.replace(/\\/g, '/'))
73
}
74
75
+// wrapper adding a few features: hidden files, onlyFiles and onlyFolders
76
export async function* dirStream(path: string, { depth=0, onlyFiles=false, onlyFolders = false }={}) {
77
if (!await isDirectory(path))
78
throw Error('ENOTDIR')
77
- const dirStream = glob.stream(depth ? '**/*' : '*', {
78
- cwd: path,
79
- dot: true,
80
- deep: depth + 1,
81
- onlyFiles,
82
- onlyDirectories: onlyFolders,
83
- suppressErrors: true,
84
- objectMode: true,
85
- unique: false,
86
- })
79
const skip = await getItemsToSkip(path)
88
- for await (const entry of dirStream) {
89
- let { path, dirent } = entry as any
90
- const isDir = dirent.isDirectory()
91
- if (!isDir && !dirent.isFile()) continue
92
- path = String(path)
93
- if (!skip?.includes(path))
94
- yield [path, isDir] as const
80
+ for await (const entry of createDirStream(path, depth)) {
81
+ const dirent = entry as DirStreamEntry
82
+ if (dirent.isDirectory() ? onlyFiles : (onlyFolders || !dirent.isFile())) continue
83
+ if (skip?.includes(entry.path)) continue
84
+ yield dirent
85
}
86
87
async function getItemsToSkip(path: string) {
src/vfs.ts
+8
-3
@@ -292,13 +292,15 @@ export async function* walkNode(parent: VfsNode, {
292
&& !masksCouldGivePermission(parent.masks, requiredPerm))
293
return
294
295
+ let n = 0
296
try {
297
let lastDir = prefixPath.slice(0, -1) || '.'
298
parentsCache.set(lastDir, parent)
298
- // it's important to keep using dirStream in deep-mode, as it is manyfold faster (it parallelizes)
299
- for await (const [path, isFolder] of dirStream(source, { depth, onlyFolders })) {
299
+ for await (const entry of dirStream(source, { depth, onlyFolders })) {
300
if (ctx?.req.aborted)
301
return
302
+ const {path} = entry
303
+ const isFolder = entry.isDirectory()
304
const name = prefixPath + (parent.rename?.[path] || path)
305
if (took?.has(normalizeFilename(name))) continue
306
if (depth) {
@@ -317,11 +319,14 @@ export async function* walkNode(parent: VfsNode, {
319
parentsCache.set(name, item)
320
if (canSee(item))
321
yield item
322
+ entry.closingBranch?.then(p =>
323
+ parentsCache.delete(p || '.'))
324
}
325
}
326
catch(e) {
323
- console.debug('glob', source, e) // ENOTDIR, or lacking permissions
327
+ console.debug('walkNode', source, e) // ENOTDIR, or lacking permissions
328
}
329
+ parentsCache.clear() // hoping for faster GC
330
331
// item will be changed, so be sure to pass a temp node
332
function canSee(item: VfsNode) {