feat(core): improve resource read & write traits (#16115)

This commit introduces two new buffer wrapper types to `deno_core`. The
main benefit of these new wrappers is that they can wrap a number of
different underlying buffer types. This allows for a more flexible read
and write API on resources that will require less copying of data
between different buffer representations.

- `BufView` is a read-only view onto a buffer. It can be backed by
`ZeroCopyBuf`, `Vec<u8>`, and `bytes::Bytes`.
- `BufViewMut` is a read-write view onto a buffer. It can be cheaply
converted into a `BufView`. It can be backed by `ZeroCopyBuf` or
`Vec<u8>`.

Both new buffer views have a cursor. This means that the start point of
the view can be constrained to write / read from just a slice of the
view. Only the start point of the slice can be adjusted. The end point
is fixed. To adjust the end point, the underlying buffer needs to be
truncated.

Readable resources have been changed to better cater to resources that
do not support BYOB reads. The basic `read` method now returns a
`BufView` instead of taking a `ZeroCopyBuf` to fill. This allows the
operation to return buffers that the resource has already allocated,
instead of forcing the caller to allocate the buffer. BYOB reads are
still very useful for resources that support them, so a new `read_byob`
method has been added that takes a `BufViewMut` to fill. `op_read`
attempts to use `read_byob` if the resource supports it, which falls
back to `read` and performs an additional copy if it does not. For
Rust->JS reads this change should have no impact, but for Rust->Rust
reads, this allows the caller to avoid an additional copy in many
scenarios. This combined with the support for `BufView` to be backed by
`bytes::Bytes` allows us to avoid one data copy when piping from a
`fetch` response into an `ext/http` response.

Writable resources have been changed to take a `BufView` instead of a
`ZeroCopyBuf` as an argument. This allows for less copying of data in
certain scenarios, as described above. Additionally a new
`Resource::write_all` method has been added that takes a `BufView` and
continually attempts to write the resource until the entire buffer has
been written. Certain resources like files can override this method to
provide a more efficient `write_all` implementation.

1 files changed, 48 insertions, 44 deletions

diff --git a/runtime/ops/io.rs b/runtime/ops/io.rs
index 18c7fb5e5..8ed6969f9 100644
--- a/runtime/ops/io.rs
+++ b/runtime/ops/io.rs
@@ -7,6 +7,8 @@ use deno_core::parking_lot::Mutex;
 use deno_core::AsyncMutFuture;
 use deno_core::AsyncRefCell;
 use deno_core::AsyncResult;
+use deno_core::BufMutView;
+use deno_core::BufView;
 use deno_core::CancelHandle;
 use deno_core::CancelTryFuture;
 use deno_core::Extension;
@@ -202,9 +204,9 @@ where
     RcRef::map(self, |r| &r.stream).borrow_mut()
   }
 
-  async fn write(self: Rc<Self>, buf: ZeroCopyBuf) -> Result<usize, AnyError> {
+  async fn write(self: Rc<Self>, data: &[u8]) -> Result<usize, AnyError> {
     let mut stream = self.borrow_mut().await;
-    let nwritten = stream.write(&buf).await?;
+    let nwritten = stream.write(data).await?;
     Ok(nwritten)
   }
 
@@ -250,16 +252,10 @@ where
     self.cancel_handle.cancel()
   }
 
-  async fn read(
-    self: Rc<Self>,
-    mut buf: ZeroCopyBuf,
-  ) -> Result<(usize, ZeroCopyBuf), AnyError> {
+  async fn read(self: Rc<Self>, data: &mut [u8]) -> Result<usize, AnyError> {
     let mut rd = self.borrow_mut().await;
-    let nread = rd
-      .read(&mut buf)
-      .try_or_cancel(self.cancel_handle())
-      .await?;
-    Ok((nread, buf))
+    let nread = rd.read(data).try_or_cancel(self.cancel_handle()).await?;
+    Ok(nread)
   }
 
   pub fn into_inner(self) -> S {
@@ -274,9 +270,7 @@ impl Resource for ChildStdinResource {
     "childStdin".into()
   }
 
-  fn write(self: Rc<Self>, buf: ZeroCopyBuf) -> AsyncResult<usize> {
-    Box::pin(self.write(buf))
-  }
+  deno_core::impl_writable!();
 
   fn shutdown(self: Rc<Self>) -> AsyncResult<()> {
     Box::pin(self.shutdown())
@@ -286,17 +280,12 @@ impl Resource for ChildStdinResource {
 pub type ChildStdoutResource = ReadOnlyResource<process::ChildStdout>;
 
 impl Resource for ChildStdoutResource {
+  deno_core::impl_readable_byob!();
+
   fn name(&self) -> Cow<str> {
     "childStdout".into()
   }
 
-  fn read_return(
-    self: Rc<Self>,
-    buf: ZeroCopyBuf,
-  ) -> AsyncResult<(usize, ZeroCopyBuf)> {
-    Box::pin(self.read(buf))
-  }
-
   fn close(self: Rc<Self>) {
     self.cancel_read_ops();
   }
@@ -305,17 +294,12 @@ impl Resource for ChildStdoutResource {
 pub type ChildStderrResource = ReadOnlyResource<process::ChildStderr>;
 
 impl Resource for ChildStderrResource {
+  deno_core::impl_readable_byob!();
+
   fn name(&self) -> Cow<str> {
     "childStderr".into()
   }
 
-  fn read_return(
-    self: Rc<Self>,
-    buf: ZeroCopyBuf,
-  ) -> AsyncResult<(usize, ZeroCopyBuf)> {
-    Box::pin(self.read(buf))
-  }
-
   fn close(self: Rc<Self>) {
     self.cancel_read_ops();
   }
@@ -534,25 +518,34 @@ impl StdFileResource {
     result
   }
 
-  async fn read(
+  async fn read_byob(
     self: Rc<Self>,
-    mut buf: ZeroCopyBuf,
-  ) -> Result<(usize, ZeroCopyBuf), AnyError> {
+    mut buf: BufMutView,
+  ) -> Result<(usize, BufMutView), AnyError> {
     self
-      .with_inner_blocking_task(
-        move |inner| -> Result<(usize, ZeroCopyBuf), AnyError> {
-          Ok((inner.read(&mut buf)?, buf))
-        },
-      )
+      .with_inner_blocking_task(move |inner| {
+        let nread = inner.read(&mut buf)?;
+        Ok((nread, buf))
+      })
       .await
   }
 
-  async fn write(self: Rc<Self>, buf: ZeroCopyBuf) -> Result<usize, AnyError> {
+  async fn write(self: Rc<Self>, data: &[u8]) -> Result<usize, AnyError> {
+    let buf = data.to_owned();
     self
       .with_inner_blocking_task(move |inner| inner.write_and_maybe_flush(&buf))
       .await
   }
 
+  async fn write_all(self: Rc<Self>, data: &[u8]) -> Result<(), AnyError> {
+    let buf = data.to_owned();
+    self
+      .with_inner_blocking_task(move |inner| {
+        inner.write_all_and_maybe_flush(&buf)
+      })
+      .await
+  }
+
   fn with_resource<F, R>(
     state: &mut OpState,
     rid: ResourceId,
@@ -641,17 +634,28 @@ impl Resource for StdFileResource {
     self.name.as_str().into()
   }
 
-  fn read_return(
-    self: Rc<Self>,
-    buf: ZeroCopyBuf,
-  ) -> AsyncResult<(usize, ZeroCopyBuf)> {
-    Box::pin(self.read(buf))
+  fn read(self: Rc<Self>, limit: usize) -> AsyncResult<deno_core::BufView> {
+    Box::pin(async move {
+      let vec = vec![0; limit];
+      let buf = BufMutView::from(vec);
+      let (nread, buf) = self.read_byob(buf).await?;
+      let mut vec = buf.unwrap_vec();
+      if vec.len() != nread {
+        vec.truncate(nread);
+      }
+      Ok(BufView::from(vec))
+    })
   }
 
-  fn write(self: Rc<Self>, buf: ZeroCopyBuf) -> AsyncResult<usize> {
-    Box::pin(self.write(buf))
+  fn read_byob(
+    self: Rc<Self>,
+    buf: deno_core::BufMutView,
+  ) -> AsyncResult<(usize, deno_core::BufMutView)> {
+    Box::pin(self.read_byob(buf))
   }
 
+  deno_core::impl_writable!(with_all);
+
   #[cfg(unix)]
   fn backing_fd(self: Rc<Self>) -> Option<std::os::unix::prelude::RawFd> {
     use std::os::unix::io::AsRawFd;