Close intermediate streams for reads/writes (#7816)

Previously temporary streams created as part of the preview1 adapter in
the wasmtime-wasi crate were left open which meant that they continued
to occupy space in the resource table and the underlying file
accidentally wasn't ever actually closed.

cc #7813

crates/test-programs/src/bin/preview1_path_open_lots.rs

@@ -0,0 +1,107 @@
+use std::{env, process};
+use test_programs::preview1::{create_file, open_scratch_directory};
+
+unsafe fn test_path_open_lots(dir_fd: wasi::Fd) {
+    create_file(dir_fd, "file");
+
+    for _ in 0..2000 {
+        let f_readonly = wasi::path_open(dir_fd, 0, "file", 0, wasi::RIGHTS_FD_READ, 0, 0)
+            .expect("open file readonly");
+
+        let buffer = &mut [0u8; 100];
+        let iovec = wasi::Iovec {
+            buf: buffer.as_mut_ptr(),
+            buf_len: buffer.len(),
+        };
+        let nread = wasi::fd_read(f_readonly, &[iovec]).expect("reading readonly file");
+        assert_eq!(nread, 0, "readonly file is empty");
+
+        wasi::fd_close(f_readonly).expect("close readonly");
+    }
+
+    for _ in 0..2000 {
+        let f_readonly = wasi::path_open(dir_fd, 0, "file", 0, wasi::RIGHTS_FD_READ, 0, 0)
+            .expect("open file readonly");
+
+        let buffer = &mut [0u8; 100];
+        let iovec = wasi::Iovec {
+            buf: buffer.as_mut_ptr(),
+            buf_len: buffer.len(),
+        };
+        let nread = wasi::fd_pread(f_readonly, &[iovec], 0).expect("reading readonly file");
+        assert_eq!(nread, 0, "readonly file is empty");
+
+        wasi::fd_close(f_readonly).expect("close readonly");
+    }
+
+    for _ in 0..2000 {
+        let f = wasi::path_open(
+            dir_fd,
+            0,
+            "file",
+            0,
+            wasi::RIGHTS_FD_READ | wasi::RIGHTS_FD_WRITE,
+            0,
+            0,
+        )
+        .unwrap();
+
+        let buffer = &[0u8; 100];
+        let ciovec = wasi::Ciovec {
+            buf: buffer.as_ptr(),
+            buf_len: buffer.len(),
+        };
+        let nwritten = wasi::fd_write(f, &[ciovec]).expect("write failed");
+        assert_eq!(nwritten, 100);
+
+        wasi::fd_close(f).unwrap();
+    }
+
+    for _ in 0..2000 {
+        let f = wasi::path_open(
+            dir_fd,
+            0,
+            "file",
+            0,
+            wasi::RIGHTS_FD_READ | wasi::RIGHTS_FD_WRITE,
+            0,
+            0,
+        )
+        .unwrap();
+
+        let buffer = &[0u8; 100];
+        let ciovec = wasi::Ciovec {
+            buf: buffer.as_ptr(),
+            buf_len: buffer.len(),
+        };
+        let nwritten = wasi::fd_pwrite(f, &[ciovec], 0).expect("write failed");
+        assert_eq!(nwritten, 100);
+
+        wasi::fd_close(f).unwrap();
+    }
+
+    wasi::path_unlink_file(dir_fd, "file").expect("removing a file");
+}
+
+fn main() {
+    let mut args = env::args();
+    let prog = args.next().unwrap();
+    let arg = if let Some(arg) = args.next() {
+        arg
+    } else {
+        eprintln!("usage: {} <scratch directory>", prog);
+        process::exit(1);
+    };
+
+    // Open scratch directory
+    let dir_fd = match open_scratch_directory(&arg) {
+        Ok(dir_fd) => dir_fd,
+        Err(err) => {
+            eprintln!("{}", err);
+            process::exit(1)
+        }
+    };
+
+    // Run the tests.
+    unsafe { test_path_open_lots(dir_fd) }
+}

crates/wasi-common/tests/all/async_.rs

@@ -283,3 +283,7 @@ async fn preview1_unicode_output() {
 async fn preview1_file_write() {
     run(PREVIEW1_FILE_WRITE, true).await.unwrap()
 }
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn preview1_path_open_lots() {
+    run(PREVIEW1_PATH_OPEN_LOTS, true).await.unwrap()
+}

crates/wasi-common/tests/all/sync.rs

@@ -269,3 +269,7 @@ fn preview1_unicode_output() {
 fn preview1_file_write() {
     run(PREVIEW1_FILE_WRITE, true).unwrap()
 }
+#[test_log::test]
+fn preview1_path_open_lots() {
+    run(PREVIEW1_PATH_OPEN_LOTS, true).unwrap()
+}

crates/wasi/src/preview2/preview1.rs

@@ -1335,8 +1335,10 @@ impl<
                         .unwrap_or_else(types::Error::trap)
                 })?;
                 let read = blocking_mode
-                    .read(self, stream, buf.len().try_into()?)
-                    .await?;
+                    .read(self, stream.borrowed(), buf.len().try_into()?)
+                    .await;
+                streams::HostInputStream::drop(self, stream).map_err(|e| types::Error::trap(e))?;
+                let read = read?;
                 let n = read.len().try_into()?;
                 let pos = pos.checked_add(n).ok_or(types::Errno::Overflow)?;
                 position.store(pos, Ordering::Relaxed);
@@ -1393,9 +1395,10 @@ impl<
                         .unwrap_or_else(types::Error::trap)
                 })?;
                 let read = blocking_mode
-                    .read(self, stream, buf.len().try_into()?)
-                    .await?;
-                (buf, read)
+                    .read(self, stream.borrowed(), buf.len().try_into()?)
+                    .await;
+                streams::HostInputStream::drop(self, stream).map_err(|e| types::Error::trap(e))?;
+                (buf, read?)
             }
             Descriptor::Stdin { .. } => {
                 // NOTE: legacy implementation returns SPIPE here
@@ -1454,7 +1457,9 @@ impl<
                     })?;
                     (stream, pos)
                 };
-                let n = blocking_mode.write(self, stream, buf).await?;
+                let n = blocking_mode.write(self, stream.borrowed(), buf).await;
+                streams::HostOutputStream::drop(self, stream).map_err(|e| types::Error::trap(e))?;
+                let n = n?;
                 if append {
                     let len = self.stat(fd2).await?;
                     position.store(len.size, Ordering::Relaxed);
@@ -1507,7 +1512,9 @@ impl<
                         .context("failed to call `write-via-stream`")
                         .unwrap_or_else(types::Error::trap)
                 })?;
-                blocking_mode.write(self, stream, buf).await?
+                let result = blocking_mode.write(self, stream.borrowed(), buf).await;
+                streams::HostOutputStream::drop(self, stream).map_err(|e| types::Error::trap(e))?;
+                result?
             }
             Descriptor::Stdout { .. } | Descriptor::Stderr { .. } => {
                 // NOTE: legacy implementation returns SPIPE here

crates/wasi/tests/all/async_.rs

@@ -281,6 +281,10 @@ async fn preview1_unicode_output() {
 async fn preview1_file_write() {
     run(PREVIEW1_FILE_WRITE_COMPONENT, false).await.unwrap()
 }
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn preview1_path_open_lots() {
+    run(PREVIEW1_PATH_OPEN_LOTS_COMPONENT, false).await.unwrap()
+}
 
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn preview2_sleep() {

crates/wasi/tests/all/preview1.rs

@@ -237,3 +237,7 @@ async fn preview1_unicode_output() {
 async fn preview1_file_write() {
     run(PREVIEW1_FILE_WRITE, true).await.unwrap()
 }
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn preview1_path_open_lots() {
+    run(PREVIEW1_PATH_OPEN_LOTS, true).await.unwrap()
+}

crates/wasi/tests/all/sync.rs

@@ -228,6 +228,10 @@ fn preview1_unicode_output() {
 fn preview1_file_write() {
     run(PREVIEW1_FILE_WRITE_COMPONENT, false).unwrap()
 }
+#[test_log::test]
+fn preview1_path_open_lots() {
+    run(PREVIEW1_PATH_OPEN_LOTS_COMPONENT, false).unwrap()
+}
 
 #[test_log::test]
 fn preview2_sleep() {