Reland "perf(http): optimize ReadableStreams backed by a resource" (#14346)

2 files changed, 23 insertions, 13 deletions

diff --git a/runtime/js/40_files.js b/runtime/js/40_files.js
index 8aa0a4972..d2148be2f 100644
--- a/runtime/js/40_files.js
+++ b/runtime/js/40_files.js
@@ -6,8 +6,8 @@
   const { read, readSync, write, writeSync } = window.__bootstrap.io;
   const { ftruncate, ftruncateSync, fstat, fstatSync } = window.__bootstrap.fs;
   const { pathFromURL } = window.__bootstrap.util;
-  const { readableStreamForRid, writableStreamForRid } =
-    window.__bootstrap.streamUtils;
+  const { writableStreamForRid } = window.__bootstrap.streamUtils;
+  const { readableStreamForRid } = window.__bootstrap.streams;
   const {
     ArrayPrototypeFilter,
     Error,
diff --git a/runtime/ops/io.rs b/runtime/ops/io.rs
index 27a48a961..34cd541d5 100644
--- a/runtime/ops/io.rs
+++ b/runtime/ops/io.rs
@@ -174,13 +174,13 @@ where
   async fn read(
     self: Rc<Self>,
     mut buf: ZeroCopyBuf,
-  ) -> Result<usize, AnyError> {
+  ) -> Result<(usize, ZeroCopyBuf), AnyError> {
     let mut rd = self.borrow_mut().await;
     let nread = rd
       .read(&mut buf)
       .try_or_cancel(self.cancel_handle())
       .await?;
-    Ok(nread)
+    Ok((nread, buf))
   }
 
   pub fn into_inner(self) -> S {
@@ -211,7 +211,10 @@ impl Resource for ChildStdoutResource {
     "childStdout".into()
   }
 
-  fn read(self: Rc<Self>, buf: ZeroCopyBuf) -> AsyncResult<usize> {
+  fn read_return(
+    self: Rc<Self>,
+    buf: ZeroCopyBuf,
+  ) -> AsyncResult<(usize, ZeroCopyBuf)> {
     Box::pin(self.read(buf))
   }
 
@@ -227,7 +230,10 @@ impl Resource for ChildStderrResource {
     "childStderr".into()
   }
 
-  fn read(self: Rc<Self>, buf: ZeroCopyBuf) -> AsyncResult<usize> {
+  fn read_return(
+    self: Rc<Self>,
+    buf: ZeroCopyBuf,
+  ) -> AsyncResult<(usize, ZeroCopyBuf)> {
     Box::pin(self.read(buf))
   }
 
@@ -271,16 +277,17 @@ impl StdFileResource {
   async fn read(
     self: Rc<Self>,
     mut buf: ZeroCopyBuf,
-  ) -> Result<usize, AnyError> {
+  ) -> Result<(usize, ZeroCopyBuf), AnyError> {
     if self.fs_file.is_some() {
       let fs_file = self.fs_file.as_ref().unwrap();
       let std_file = fs_file.0.as_ref().unwrap().clone();
-      tokio::task::spawn_blocking(move || {
-        let mut std_file = std_file.lock().unwrap();
-        std_file.read(&mut buf)
-      })
+      tokio::task::spawn_blocking(
+        move || -> Result<(usize, ZeroCopyBuf), AnyError> {
+          let mut std_file = std_file.lock().unwrap();
+          Ok((std_file.read(&mut buf)?, buf))
+        },
+      )
       .await?
-      .map_err(AnyError::from)
     } else {
       Err(resource_unavailable())
     }
@@ -330,7 +337,10 @@ impl Resource for StdFileResource {
     self.name.as_str().into()
   }
 
-  fn read(self: Rc<Self>, buf: ZeroCopyBuf) -> AsyncResult<usize> {
+  fn read_return(
+    self: Rc<Self>,
+    buf: ZeroCopyBuf,
+  ) -> AsyncResult<(usize, ZeroCopyBuf)> {
     Box::pin(self.read(buf))
   }