From 958f4c36a83043bfb196662f768fcf585e2977d9 Mon Sep 17 00:00:00 2001 From: Robert Nagy Date: Sat, 8 Apr 2023 19:15:59 +0200 Subject: [PATCH 1/5] fs: fs stream should not wait for handle with refs --- lib/internal/fs/promises.js | 1 + lib/internal/fs/streams.js | 21 +++++++++++++++------ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index 41c17de95ed3..ad858cfdf405 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -980,6 +980,7 @@ module.exports = { }, FileHandle, + kRefs, kRef, kUnref, }; diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js index f75d0fba9172..ef2a44a252dd 100644 --- a/lib/internal/fs/streams.js +++ b/lib/internal/fs/streams.js @@ -26,7 +26,7 @@ const { } = require('internal/validators'); const { errorOrDestroy } = require('internal/streams/destroy'); const fs = require('fs'); -const { kRef, kUnref, FileHandle } = require('internal/fs/promises'); +const { kRef, kUnref, kRefs, FileHandle } = require('internal/fs/promises'); const { Buffer } = require('buffer'); const { copyObject, @@ -36,6 +36,8 @@ const { } = require('internal/fs/utils'); const { Readable, Writable, finished } = require('stream'); const { toPathIfFileURL } = require('internal/url'); +const assert = require('internal/assert'); + const kIoDone = Symbol('kIoDone'); const kIsPerformingIO = Symbol('kIsPerformingIO'); @@ -80,15 +82,21 @@ function _construct(callback) { } // This generates an fs operations structure for a FileHandle -const FileHandleOperations = (handle) => { +const FileHandleOperations = (handle, closeHandler) => { return { open: (path, flags, mode, cb) => { throw new ERR_METHOD_NOT_IMPLEMENTED('open()'); }, close: (fd, cb) => { + handle.off('close', closeHandler); handle[kUnref](); - PromisePrototypeThen(handle.close(), - () => cb(), cb); + if (handle[kRefs] > 0) { + // Someone else has a ref to the handle. + process.nextTick(cb); + } else { + PromisePrototypeThen(handle.close(), + () => cb(), cb); + } }, read: (fd, buf, offset, length, pos, cb) => { PromisePrototypeThen(handle.read(buf, offset, length, pos), @@ -134,10 +142,11 @@ function importFd(stream, options) { // FileHandle is not supported with custom fs operations throw new ERR_METHOD_NOT_IMPLEMENTED('FileHandle with fs'); } + const closeHandler = FunctionPrototypeBind(stream.close, stream); stream[kHandle] = options.fd; - stream[kFs] = FileHandleOperations(stream[kHandle]); + stream[kFs] = FileHandleOperations(stream[kHandle], closeHandler); stream[kHandle][kRef](); - options.fd.on('close', FunctionPrototypeBind(stream.close, stream)); + stream[kHandle].on('close', closeHandler); return options.fd.fd; } From 2af786492a08cc2bb3183dee0ae96696026b3940 Mon Sep 17 00:00:00 2001 From: Robert Nagy Date: Sat, 8 Apr 2023 19:12:07 +0200 Subject: [PATCH 2/5] fs: emit close after handle close --- lib/internal/fs/promises.js | 7 +++++-- lib/internal/fs/streams.js | 3 ++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index ad858cfdf405..2a712903ba7d 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -225,7 +225,10 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { this[kFd] = -1; this[kClosePromise] = SafePromisePrototypeFinally( this[kHandle].close(), - () => { this[kClosePromise] = undefined; }, + () => { + this[kClosePromise] = undefined; + this.emit('close'); + } ); } else { this[kClosePromise] = SafePromisePrototypeFinally( @@ -236,11 +239,11 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { this[kClosePromise] = undefined; this[kCloseReject] = undefined; this[kCloseResolve] = undefined; + this.emit('close'); }, ); } - this.emit('close'); return this[kClosePromise]; }; diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js index ef2a44a252dd..fb13904db443 100644 --- a/lib/internal/fs/streams.js +++ b/lib/internal/fs/streams.js @@ -90,7 +90,7 @@ const FileHandleOperations = (handle, closeHandler) => { close: (fd, cb) => { handle.off('close', closeHandler); handle[kUnref](); - if (handle[kRefs] > 0) { + if (handle[kRefs] > 1) { // Someone else has a ref to the handle. process.nextTick(cb); } else { @@ -142,6 +142,7 @@ function importFd(stream, options) { // FileHandle is not supported with custom fs operations throw new ERR_METHOD_NOT_IMPLEMENTED('FileHandle with fs'); } + assert(stream[kHandle][kRefs] > 0); const closeHandler = FunctionPrototypeBind(stream.close, stream); stream[kHandle] = options.fd; stream[kFs] = FileHandleOperations(stream[kHandle], closeHandler); From 049703d656c135e39935fbca76d5daf25d18259b Mon Sep 17 00:00:00 2001 From: Robert Nagy Date: Sat, 8 Apr 2023 19:12:34 +0200 Subject: [PATCH 3/5] fs: add some sanity assertions --- lib/internal/fs/promises.js | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index 2a712903ba7d..260890b41440 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -340,10 +340,12 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { } [kRef]() { + assert(this[kFd] !== -1); this[kRefs]++; } [kUnref]() { + assert(this[kFd] !== -1); this[kRefs]--; if (this[kRefs] === 0) { this[kFd] = -1; From aff3657776c1b53e44c03eec9d2bd4d142c31cde Mon Sep 17 00:00:00 2001 From: Robert Nagy Date: Sat, 8 Apr 2023 19:35:31 +0200 Subject: [PATCH 4/5] fs: handle close --- lib/internal/fs/promises.js | 53 ++++++++++++++++++------------------- 1 file changed, 26 insertions(+), 27 deletions(-) diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index 260890b41440..6f5d2921a592 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -100,6 +100,7 @@ const kRefs = Symbol('kRefs'); const kClosePromise = Symbol('kClosePromise'); const kCloseResolve = Symbol('kCloseResolve'); const kCloseReject = Symbol('kCloseReject'); +const kClosing = Symbol('kClosing'); const kRef = Symbol('kRef'); const kUnref = Symbol('kUnref'); const kLocked = Symbol('kLocked'); @@ -137,7 +138,17 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { this[kFd] = filehandle ? filehandle.fd : -1; this[kRefs] = 1; - this[kClosePromise] = null; + this[kClosing] = false; + this[kClosePromise] = SafePromisePrototypeFinally( + new Promise((resolve, reject) => { + this[kCloseResolve] = resolve; + this[kCloseReject] = reject; + }), () => { + this[kClosePromise] = undefined; + this[kCloseReject] = undefined; + this[kCloseResolve] = undefined; + }, + ); } getAsyncId() { @@ -216,33 +227,20 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { return PromiseResolve(); } - if (this[kClosePromise]) { + if (this[kClosing]) { return this[kClosePromise]; } - this[kRefs]--; - if (this[kRefs] === 0) { - this[kFd] = -1; - this[kClosePromise] = SafePromisePrototypeFinally( - this[kHandle].close(), - () => { - this[kClosePromise] = undefined; - this.emit('close'); - } - ); - } else { - this[kClosePromise] = SafePromisePrototypeFinally( - new Promise((resolve, reject) => { - this[kCloseResolve] = resolve; - this[kCloseReject] = reject; - }), () => { - this[kClosePromise] = undefined; - this[kCloseReject] = undefined; - this[kCloseResolve] = undefined; - this.emit('close'); - }, - ); - } + this[kClosing] = true; + this[kUnref](); + + SafePromisePrototypeFinally( + this[kClosePromise], + () => { + this[kClosePromise] = undefined; + this.emit('close'); + } + ); return this[kClosePromise]; }; @@ -255,7 +253,7 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { readableWebStream() { if (this[kFd] === -1) throw new ERR_INVALID_STATE('The FileHandle is closed'); - if (this[kClosePromise]) + if (this[kClosing]) throw new ERR_INVALID_STATE('The FileHandle is closing'); if (this[kLocked]) throw new ERR_INVALID_STATE('The FileHandle is locked'); @@ -314,7 +312,7 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { } [kTransfer]() { - if (this[kClosePromise] || this[kRefs] > 1) { + if (this[kClosing] || this[kRefs] > 1) { throw lazyDOMException('Cannot transfer FileHandle while in use', 'DataCloneError'); } @@ -347,6 +345,7 @@ class FileHandle extends EventEmitterMixin(JSTransferable) { [kUnref]() { assert(this[kFd] !== -1); this[kRefs]--; + if (this[kRefs] === 0) { this[kFd] = -1; PromisePrototypeThen( From 42fd2a5840f4f49a9f2f7169474ea50136d3752a Mon Sep 17 00:00:00 2001 From: Robert Nagy Date: Sat, 8 Apr 2023 19:47:00 +0200 Subject: [PATCH 5/5] fixup! fs: fs stream should not wait for handle with refs --- lib/internal/fs/streams.js | 2 +- test/parallel/test-fs-read-stream-file-handle.js | 12 +++++++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js index fb13904db443..a5c300049fdf 100644 --- a/lib/internal/fs/streams.js +++ b/lib/internal/fs/streams.js @@ -142,7 +142,7 @@ function importFd(stream, options) { // FileHandle is not supported with custom fs operations throw new ERR_METHOD_NOT_IMPLEMENTED('FileHandle with fs'); } - assert(stream[kHandle][kRefs] > 0); + assert(options.fd[kRefs] > 0); const closeHandler = FunctionPrototypeBind(stream.close, stream); stream[kHandle] = options.fd; stream[kFs] = FileHandleOperations(stream[kHandle], closeHandler); diff --git a/test/parallel/test-fs-read-stream-file-handle.js b/test/parallel/test-fs-read-stream-file-handle.js index 810745149dc5..2560b7e07f7a 100644 --- a/test/parallel/test-fs-read-stream-file-handle.js +++ b/test/parallel/test-fs-read-stream-file-handle.js @@ -29,12 +29,22 @@ fs.promises.open(file, 'r').then((handle) => { fs.promises.open(file, 'r').then((handle) => { handle.on('close', common.mustCall()); const stream = fs.createReadStream(null, { fd: handle }); - stream.on('data', common.mustNotCall()); + stream.on('data', common.mustCall(() => { + stream.destroy(); + })); stream.on('close', common.mustCall()); return handle.close(); }).then(common.mustCall()); +fs.promises.open(file, 'r').then((handle) => { + handle.on('close', common.mustCall()); + const stream = fs.createReadStream(null, { fd: handle }); + stream.on('data', common.mustNotCall()); + stream.on('close', common.mustCall()); + stream.destroy(); +}).then(common.mustCall()); + fs.promises.open(file, 'r').then((handle) => { handle.on('close', common.mustCall()); const stream = fs.createReadStream(null, { fd: handle });