Clean up helper task in AsyncReadStream faster (#13105)

The previous implementation would hit EOF on the input object, then hit
EOF again, and then block on sending a value over the channel. There's
no need for the task to stick around and infinitely generate EOF
messages, however, so exit the task as soon as the first EOF is hit and
a message is sent.

Author: alexcrichton

crates/wasi/src/p2/pipe.rs

@@ -165,7 +165,10 @@ impl AsyncReadStream {
                 use tokio::io::AsyncReadExt;
                 let mut buf = bytes::BytesMut::with_capacity(crate::MAX_READ_SIZE_ALLOC);
                 let sent = match reader.read_buf(&mut buf).await {
-                    Ok(nbytes) if nbytes == 0 => sender.send(Err(StreamError::Closed)).await,
+                    Ok(nbytes) if nbytes == 0 => {
+                        let _ = sender.send(Err(StreamError::Closed)).await;
+                        break;
+                    }
                     Ok(_) => sender.send(Ok(buf.freeze())).await,
                     Err(e) => {
                         sender
@@ -239,15 +242,12 @@ impl InputStream for AsyncReadStream {
                 self.closed = true;
                 Err(e)
             }
-            Err(TryRecvError::Empty) => {
-                if self.closed {
-                    // Note: if the stream is already closed it should return an error,
-                    //       returning empty list would break the wasi contract (returning 0 and ready)
-                    Err(StreamError::Closed)
-                } else {
-                    Ok(Bytes::new())
-                }
+            // Note: if the stream is already closed it should return an error,
+            //       returning empty list would break the wasi contract (returning 0 and ready)
+            Err(TryRecvError::Empty | TryRecvError::Disconnected) if self.closed => {
+                Err(StreamError::Closed)
             }
+            Err(TryRecvError::Empty) => Ok(Bytes::new()),
             Err(TryRecvError::Disconnected) => Err(StreamError::Trap(format_err!(
                 "AsyncReadStream sender died - should be impossible"
             ))),