fs: synchronize close with other I/O for streams

Part of the flakiness in the
parallel/test-readline-async-iterators-destroy test comes from
fs streams starting `_read()` and `_destroy()` without waiting
for the other to finish, which can lead to the `fs.read()` call
resulting in `EBADF` if timing is bad.
Fix this by synchronizing write and read operations with `close()`.
Refs: #30660
PR-URL: #30837
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

Author: addaleax

lib/internal/fs/streams.js

@@ -7,10 +7,12 @@ const {
  NumberIsSafeInteger,
  ObjectDefineProperty,
  ObjectSetPrototypeOf,
+ Symbol,
 }=primordials;
 
 const{
-ERR_OUT_OF_RANGE
+ERR_OUT_OF_RANGE,
+ERR_STREAM_DESTROYED
 }=require('internal/errors').codes;
 const{ validateNumber }=require('internal/validators');
 constfs=require('fs');
@@ -21,6 +23,8 @@ const {
 }=require('internal/fs/utils');
 const{ Readable, Writable }=require('stream');
 const{ toPathIfFileURL }=require('internal/url');
+constkIoDone=Symbol('kIoDone');
+constkIsPerformingIO=Symbol('kIsPerformingIO');
 
 constkMinPoolSpace=128;
 
@@ -85,6 +89,7 @@ function ReadStream(path, options) {
 this.pos=undefined;
 this.bytesRead=0;
 this.closed=false;
+this[kIsPerformingIO]=false;
 
 if(this.start!==undefined){
 checkPosition(this.start,'start');
@@ -143,6 +148,8 @@ ReadStream.prototype._read = function(n) {
 });
 }
 
+if(this.destroyed)return;
+
 if(!pool||pool.length-pool.used<kMinPoolSpace){
 // Discard the old pool.
 allocNewPool(this.readableHighWaterMark);
@@ -166,7 +173,12 @@ ReadStream.prototype._read = function(n) {
 returnthis.push(null);
 
 // the actual read.
+this[kIsPerformingIO]=true;
 fs.read(this.fd,pool,pool.used,toRead,this.pos,(er,bytesRead)=>{
+this[kIsPerformingIO]=false;
+// Tell ._destroy() that it's safe to close the fd now.
+if(this.destroyed)returnthis.emit(kIoDone,er);
+
 if(er){
 if(this.autoClose){
 this.destroy();
@@ -212,8 +224,12 @@ ReadStream.prototype._destroy = function(err, cb) {
 return;
 }
 
+if(this[kIsPerformingIO]){
+this.once(kIoDone,(er)=>closeFsStream(this,cb,err||er));
+return;
+}
+
 closeFsStream(this,cb,err);
-this.fd=null;
 };
 
 functioncloseFsStream(stream,cb,err){
@@ -224,6 +240,8 @@ function closeFsStream(stream, cb, err) {
 if(!er)
 stream.emit('close');
 });
+
+stream.fd=null;
 }
 
 ReadStream.prototype.close=function(cb){
@@ -262,6 +280,7 @@ function WriteStream(path, options) {
 this.pos=undefined;
 this.bytesWritten=0;
 this.closed=false;
+this[kIsPerformingIO]=false;
 
 if(this.start!==undefined){
 checkPosition(this.start,'start');
@@ -316,7 +335,17 @@ WriteStream.prototype._write = function(data, encoding, cb) {
 });
 }
 
+if(this.destroyed)returncb(newERR_STREAM_DESTROYED('write'));
+
+this[kIsPerformingIO]=true;
 fs.write(this.fd,data,0,data.length,this.pos,(er,bytes)=>{
+this[kIsPerformingIO]=false;
+// Tell ._destroy() that it's safe to close the fd now.
+if(this.destroyed){
+cb(er);
+returnthis.emit(kIoDone,er);
+}
+
 if(er){
 if(this.autoClose){
 this.destroy();
@@ -339,7 +368,8 @@ WriteStream.prototype._writev = function(data, cb) {
 });
 }
 
-constself=this;
+if(this.destroyed)returncb(newERR_STREAM_DESTROYED('write'));
+
 constlen=data.length;
 constchunks=newArray(len);
 letsize=0;
@@ -351,12 +381,22 @@ WriteStream.prototype._writev = function(data, cb) {
 size+=chunk.length;
 }
 
-fs.writev(this.fd,chunks,this.pos,function(er,bytes){
+this[kIsPerformingIO]=true;
+fs.writev(this.fd,chunks,this.pos,(er,bytes)=>{
+this[kIsPerformingIO]=false;
+// Tell ._destroy() that it's safe to close the fd now.
+if(this.destroyed){
+cb(er);
+returnthis.emit(kIoDone,er);
+}
+
 if(er){
-self.destroy();
+if(this.autoClose){
+this.destroy();
+}
 returncb(er);
 }
-self.bytesWritten+=bytes;
+this.bytesWritten+=bytes;
 cb();
 });