@@ -7,10 +7,12 @@ const {
7
7
NumberIsSafeInteger,
8
8
ObjectDefineProperty,
9
9
ObjectSetPrototypeOf,
10
+ Symbol,
10
11
} = primordials ;
11
12
12
13
const {
13
- ERR_OUT_OF_RANGE
14
+ ERR_OUT_OF_RANGE ,
15
+ ERR_STREAM_DESTROYED
14
16
} = require ( 'internal/errors' ) . codes ;
15
17
const { validateNumber } = require ( 'internal/validators' ) ;
16
18
const fs = require ( 'fs' ) ;
@@ -21,6 +23,8 @@ const {
21
23
} = require ( 'internal/fs/utils' ) ;
22
24
const { Readable, Writable } = require ( 'stream' ) ;
23
25
const { toPathIfFileURL } = require ( 'internal/url' ) ;
26
+ const kIoDone = Symbol ( 'kIoDone' ) ;
27
+ const kIsPerformingIO = Symbol ( 'kIsPerformingIO' ) ;
24
28
25
29
const kMinPoolSpace = 128 ;
26
30
@@ -85,6 +89,7 @@ function ReadStream(path, options) {
85
89
this . pos = undefined ;
86
90
this . bytesRead = 0 ;
87
91
this . closed = false ;
92
+ this [ kIsPerformingIO ] = false ;
88
93
89
94
if ( this . start !== undefined ) {
90
95
checkPosition ( this . start , 'start' ) ;
@@ -143,6 +148,8 @@ ReadStream.prototype._read = function(n) {
143
148
} ) ;
144
149
}
145
150
151
+ if ( this . destroyed ) return ;
152
+
146
153
if ( ! pool || pool . length - pool . used < kMinPoolSpace ) {
147
154
// Discard the old pool.
148
155
allocNewPool ( this . readableHighWaterMark ) ;
@@ -166,7 +173,12 @@ ReadStream.prototype._read = function(n) {
166
173
return this . push ( null ) ;
167
174
168
175
// the actual read.
176
+ this [ kIsPerformingIO ] = true ;
169
177
fs . read ( this . fd , pool , pool . used , toRead , this . pos , ( er , bytesRead ) => {
178
+ this [ kIsPerformingIO ] = false ;
179
+ // Tell ._destroy() that it's safe to close the fd now.
180
+ if ( this . destroyed ) return this . emit ( kIoDone , er ) ;
181
+
170
182
if ( er ) {
171
183
if ( this . autoClose ) {
172
184
this . destroy ( ) ;
@@ -212,8 +224,12 @@ ReadStream.prototype._destroy = function(err, cb) {
212
224
return ;
213
225
}
214
226
227
+ if ( this [ kIsPerformingIO ] ) {
228
+ this . once ( kIoDone , ( er ) => closeFsStream ( this , cb , err || er ) ) ;
229
+ return ;
230
+ }
231
+
215
232
closeFsStream ( this , cb , err ) ;
216
- this . fd = null ;
217
233
} ;
218
234
219
235
function closeFsStream ( stream , cb , err ) {
@@ -224,6 +240,8 @@ function closeFsStream(stream, cb, err) {
224
240
if ( ! er )
225
241
stream . emit ( 'close' ) ;
226
242
} ) ;
243
+
244
+ stream . fd = null ;
227
245
}
228
246
229
247
ReadStream . prototype . close = function ( cb ) {
@@ -262,6 +280,7 @@ function WriteStream(path, options) {
262
280
this . pos = undefined ;
263
281
this . bytesWritten = 0 ;
264
282
this . closed = false ;
283
+ this [ kIsPerformingIO ] = false ;
265
284
266
285
if ( this . start !== undefined ) {
267
286
checkPosition ( this . start , 'start' ) ;
@@ -316,7 +335,17 @@ WriteStream.prototype._write = function(data, encoding, cb) {
316
335
} ) ;
317
336
}
318
337
338
+ if ( this . destroyed ) return cb ( new ERR_STREAM_DESTROYED ( 'write' ) ) ;
339
+
340
+ this [ kIsPerformingIO ] = true ;
319
341
fs . write ( this . fd , data , 0 , data . length , this . pos , ( er , bytes ) => {
342
+ this [ kIsPerformingIO ] = false ;
343
+ // Tell ._destroy() that it's safe to close the fd now.
344
+ if ( this . destroyed ) {
345
+ cb ( er ) ;
346
+ return this . emit ( kIoDone , er ) ;
347
+ }
348
+
320
349
if ( er ) {
321
350
if ( this . autoClose ) {
322
351
this . destroy ( ) ;
@@ -339,7 +368,8 @@ WriteStream.prototype._writev = function(data, cb) {
339
368
} ) ;
340
369
}
341
370
342
- const self = this ;
371
+ if ( this . destroyed ) return cb ( new ERR_STREAM_DESTROYED ( 'write' ) ) ;
372
+
343
373
const len = data . length ;
344
374
const chunks = new Array ( len ) ;
345
375
let size = 0 ;
@@ -351,12 +381,22 @@ WriteStream.prototype._writev = function(data, cb) {
351
381
size += chunk . length ;
352
382
}
353
383
354
- fs . writev ( this . fd , chunks , this . pos , function ( er , bytes ) {
384
+ this [ kIsPerformingIO ] = true ;
385
+ fs . writev ( this . fd , chunks , this . pos , ( er , bytes ) => {
386
+ this [ kIsPerformingIO ] = false ;
387
+ // Tell ._destroy() that it's safe to close the fd now.
388
+ if ( this . destroyed ) {
389
+ cb ( er ) ;
390
+ return this . emit ( kIoDone , er ) ;
391
+ }
392
+
355
393
if ( er ) {
356
- self . destroy ( ) ;
394
+ if ( this . autoClose ) {
395
+ this . destroy ( ) ;
396
+ }
357
397
return cb ( er ) ;
358
398
}
359
- self . bytesWritten += bytes ;
399
+ this . bytesWritten += bytes ;
360
400
cb ( ) ;
361
401
} ) ;
362
402
0 commit comments