refactor Stream. merge IoVec from Linux and Zircon

kernel-hal/src/lib.rs

@@ -43,7 +43,6 @@ mod context;
 mod dummy;
 mod future;
 pub mod user;
-pub mod user_io_vec;
 pub mod vdso;
 
 pub use self::context::*;

kernel-hal/src/user.rs

@@ -2,6 +2,7 @@ use alloc::string::String;
 use alloc::vec::Vec;
 use core::fmt::{Debug, Formatter};
 use core::marker::PhantomData;
+use core::ops::{Deref, DerefMut};
 
 #[repr(C)]
 pub struct UserPtr<T, P: Policy> {
@@ -36,6 +37,8 @@ pub enum Error {
     InvalidUtf8,
     InvalidPointer,
     BufferTooSmall,
+    InvalidLength,
+    InvalidVectorAddress,
 }
 
 impl<T, P: Policy> Debug for UserPtr<T, P> {
@@ -193,3 +196,121 @@ impl<P: Write> UserPtr<u8, P> {
         Ok(())
     }
 }
+
+#[derive(Debug)]
+#[repr(C)]
+pub struct IoVec<P: Policy> {
+    /// Starting address
+    ptr: UserPtr<u8, P>,
+    /// Number of bytes to transfer
+    len: usize,
+}
+
+pub type IoVecIn = IoVec<In>;
+pub type IoVecOut = IoVec<Out>;
+
+/// A valid IoVecs request from user
+#[derive(Debug)]
+pub struct IoVecs<P: Policy> {
+    vec: Vec<IoVec<P>>,
+}
+
+impl<P: Policy> UserInPtr<IoVec<P>> {
+    pub fn read_iovecs(&self, count: usize) -> Result<IoVecs<P>> {
+        if self.ptr.is_null() {
+            return Err(Error::InvalidPointer);
+        }
+        let vec = self.read_array(count)?;
+        // The sum of length should not overflow.
+        let mut total_count = 0usize;
+        for io_vec in vec.iter() {
+            let (result, overflow) = total_count.overflowing_add(io_vec.len());
+            if overflow {
+                return Err(Error::InvalidLength);
+            }
+            total_count = result;
+        }
+        Ok(IoVecs { vec })
+    }
+}
+
+impl<P: Policy> IoVecs<P> {
+    pub fn total_len(&self) -> usize {
+        self.vec.iter().map(|vec| vec.len).sum()
+    }
+}
+
+impl<P: Read> IoVecs<P> {
+    pub fn read_to_vec(&self) -> Result<Vec<u8>> {
+        let mut buf = Vec::new();
+        for vec in self.vec.iter() {
+            buf.extend(vec.ptr.read_array(vec.len)?);
+        }
+        Ok(buf)
+    }
+}
+
+impl<P: Write> IoVecs<P> {
+    pub fn write_from_buf(&mut self, mut buf: &[u8]) -> Result<usize> {
+        let buf_len = buf.len();
+        for vec in self.vec.iter_mut() {
+            let copy_len = vec.len.min(buf.len());
+            if copy_len == 0 {
+                continue;
+            }
+            vec.ptr.write_array(&buf[..copy_len])?;
+            buf = &buf[copy_len..];
+        }
+        Ok(buf_len - buf.len())
+    }
+}
+
+impl<P: Policy> Deref for IoVecs<P> {
+    type Target = [IoVec<P>];
+
+    fn deref(&self) -> &Self::Target {
+        self.vec.as_slice()
+    }
+}
+
+impl<P: Write> DerefMut for IoVecs<P> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.vec.as_mut_slice()
+    }
+}
+
+impl<P: Policy> IoVec<P> {
+    pub fn is_null(&self) -> bool {
+        self.ptr.is_null()
+    }
+
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+
+    pub fn check(&self) -> Result<()> {
+        self.ptr.check()
+    }
+
+    pub fn as_slice(&self) -> Result<&[u8]> {
+        if self.ptr.is_null() {
+            return Err(Error::InvalidVectorAddress);
+        }
+        let slice = unsafe { core::slice::from_raw_parts(self.ptr.as_ptr(), self.len) };
+        Ok(slice)
+    }
+}
+
+impl<P: Write> IoVec<P> {
+    pub fn as_mut_slice(&mut self) -> Result<&mut [u8]> {
+        if self.ptr.is_null() {
+            return Err(Error::InvalidVectorAddress);
+        }
+        let slice = unsafe { core::slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len) };
+        Ok(slice)
+    }
+}

kernel-hal/src/user_io_vec.rs

@@ -1,75 +0,0 @@
-use super::user::*;
-use core::slice;
-
-type VecResult<T> = core::result::Result<T, VecError>;
-
-const MAX_LENGTH: usize = 0x1000;
-
-#[repr(C)]
-pub struct IoVec<T, P: Policy> {
-    ptr: UserPtr<T, P>,
-    len: usize,
-}
-
-pub type InIoVec<T> = IoVec<T, In>;
-pub type OutIoVec<T> = IoVec<T, Out>;
-
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-pub enum VecError {
-    PtrErr(Error),
-    LengthErr,
-}
-
-impl From<Error> for VecError {
-    fn from(err: Error) -> VecError {
-        VecError::PtrErr(err)
-    }
-}
-
-impl<T, P: Policy> IoVec<T, P> {
-    pub fn is_null(&self) -> bool {
-        self.ptr.is_null()
-    }
-
-    pub fn len(&self) -> usize {
-        self.len
-    }
-
-    pub fn is_empty(&self) -> bool {
-        self.len == 0
-    }
-
-    pub fn as_user_ptr(&self) -> UserPtr<T, P> {
-        unimplemented!()
-    }
-
-    pub fn as_ptr(&self) -> *const T {
-        self.ptr.as_ptr()
-    }
-
-    pub fn check(&self) -> VecResult<()> {
-        self.ptr.check()?;
-        if self.len > MAX_LENGTH {
-            return Err(VecError::LengthErr);
-        }
-        Ok(())
-    }
-
-    pub fn as_slice(&self) -> VecResult<&[T]> {
-        self.check()?;
-        let slice = unsafe { slice::from_raw_parts(self.ptr.as_ptr(), self.len) };
-        Ok(slice)
-    }
-}
-
-impl<T, P: Write> IoVec<T, P> {
-    pub fn as_mut_slice(&self) -> VecResult<&mut [T]> {
-        self.check()?;
-        let slice = unsafe { slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len) };
-        Ok(slice)
-    }
-
-    pub fn as_mut_ptr(&self) -> *mut T {
-        self.ptr.as_ptr()
-    }
-}

linux-object/src/error.rs

@@ -166,6 +166,8 @@ impl From<Error> for LxError {
             Error::InvalidUtf8 => LxError::EINVAL,
             Error::InvalidPointer => LxError::EFAULT,
             Error::BufferTooSmall => LxError::ENOBUFS,
+            Error::InvalidLength => LxError::EINVAL,
+            Error::InvalidVectorAddress => LxError::EINVAL,
         }
     }
 }

linux-syscall/src/file/file.rs

@@ -71,11 +71,11 @@ impl Syscall<'_> {
     pub fn sys_readv(
         &self,
         fd: FileDesc,
-        iov_ptr: UserInPtr<IoVec<Out>>,
+        iov_ptr: UserInPtr<IoVecOut>,
         iov_count: usize,
     ) -> SysResult {
         info!("readv: fd={:?}, iov={:?}, count={}", fd, iov_ptr, iov_count);
-        let mut iovs = IoVecs::new(iov_ptr, iov_count)?;
+        let mut iovs = iov_ptr.read_iovecs(iov_count)?;
         let proc = self.linux_process();
         let file_like = proc.get_file_like(fd)?;
         let mut buf = vec![0u8; iovs.total_len()];
@@ -87,14 +87,14 @@ impl Syscall<'_> {
     pub fn sys_writev(
         &self,
         fd: FileDesc,
-        iov_ptr: UserInPtr<IoVec<In>>,
+        iov_ptr: UserInPtr<IoVecIn>,
         iov_count: usize,
     ) -> SysResult {
         info!(
             "writev: fd={:?}, iov={:?}, count={}",
             fd, iov_ptr, iov_count
         );
-        let iovs = IoVecs::new(iov_ptr, iov_count)?;
+        let iovs = iov_ptr.read_iovecs(iov_count)?;
         let buf = iovs.read_to_vec()?;
         let proc = self.linux_process();
         let file_like = proc.get_file_like(fd)?;

linux-syscall/src/lib.rs

@@ -11,7 +11,7 @@ extern crate alloc;
 extern crate log;
 
 use {
-    self::{consts::SyscallType as Sys, util::*},
+    self::consts::SyscallType as Sys,
     alloc::sync::Arc,
     core::convert::TryFrom,
     kernel_hal::{user::*, GeneralRegs},
@@ -24,7 +24,6 @@ mod file;
 mod misc;
 mod task;
 mod time;
-mod util;
 mod vm;
 
 pub struct Syscall<'a> {

linux-syscall/src/util.rs

@@ -1,55 +0,0 @@
-use alloc::vec::Vec;
-use kernel_hal::user::*;
-use linux_object::error::LxResult;
-
-#[derive(Debug)]
-#[repr(C)]
-pub struct IoVec<P: Policy> {
-    /// Starting address
-    base: UserPtr<u8, P>,
-    /// Number of bytes to transfer
-    len: usize,
-}
-
-/// A valid IoVecs request from user
-#[derive(Debug)]
-pub struct IoVecs<P: Policy> {
-    vec: Vec<IoVec<P>>,
-}
-
-impl<P: Policy> IoVecs<P> {
-    pub fn new(ptr: UserInPtr<IoVec<P>>, count: usize) -> LxResult<Self> {
-        Ok(IoVecs {
-            vec: ptr.read_array(count)?,
-        })
-    }
-
-    pub fn total_len(&self) -> usize {
-        self.vec.iter().map(|vec| vec.len).sum()
-    }
-}
-
-impl<P: Read> IoVecs<P> {
-    pub fn read_to_vec(&self) -> LxResult<Vec<u8>> {
-        let mut buf = Vec::new();
-        for vec in self.vec.iter() {
-            buf.extend(vec.base.read_array(vec.len)?);
-        }
-        Ok(buf)
-    }
-}
-
-impl<P: Write> IoVecs<P> {
-    pub fn write_from_buf(&mut self, mut buf: &[u8]) -> LxResult<usize> {
-        let buf_len = buf.len();
-        for vec in self.vec.iter_mut() {
-            let copy_len = vec.len.min(buf.len());
-            if copy_len == 0 {
-                continue;
-            }
-            vec.base.write_array(&buf[..copy_len])?;
-            buf = &buf[copy_len..];
-        }
-        Ok(buf_len - buf.len())
-    }
-}

zircon-object/src/error.rs

@@ -235,17 +235,8 @@ impl From<Error> for ZxError {
             Error::InvalidUtf8 => ZxError::INVALID_ARGS,
             Error::InvalidPointer => ZxError::INVALID_ARGS,
             Error::BufferTooSmall => ZxError::BUFFER_TOO_SMALL,
-        }
+            Error::InvalidLength => ZxError::INVALID_ARGS,
-    }
+            Error::InvalidVectorAddress => ZxError::NOT_FOUND,
-}
-
-use kernel_hal::user_io_vec::VecError;
-
-impl From<VecError> for ZxError {
-    fn from(e: VecError) -> Self {
-        match e {
-            VecError::PtrErr(err) => ZxError::from(err),
-            VecError::LengthErr => ZxError::INVALID_ARGS,
         }
     }
 }

zircon-object/src/vm/stream.rs

@@ -1,7 +1,4 @@
-use {
+use {super::*, crate::object::*, alloc::sync::Arc, numeric_enum_macro::numeric_enum, spin::Mutex};
-    super::*, crate::object::*, alloc::sync::Arc, bitflags::bitflags, kernel_hal::user_io_vec::*,
-    numeric_enum_macro::numeric_enum, spin::Mutex,
-};
 
 /// A readable, writable, seekable interface to some underlying storage
 ///
@@ -11,27 +8,13 @@ use {
 /// storage, typically a VMO.
 pub struct Stream {
     base: KObjectBase,
-    options: StreamOptions,
+    options: u32,
     vmo: Arc<VmObject>,
     seek: Mutex<usize>,
 }
 
 impl_kobject!(Stream);
 
-bitflags! {
-    #[derive(Default)]
-    pub struct StreamOptions: u32 {
-        #[allow(clippy::identity_op)]
-        // These can be passed to stream_create()
-        const STREAM_MODE_READ           = 1;
-        const STREAM_MODE_WRITE          = 1 << 1;
-        const STREAM_CREATE_MASK         = Self::STREAM_MODE_READ.bits | Self::STREAM_MODE_WRITE.bits;
-
-        // These can be passed to stream_writev()
-        const STREAM_APPEND              = 1;
-    }
-}
-
 numeric_enum! {
     #[repr(usize)]
     #[derive(Debug)]
@@ -44,40 +27,17 @@ numeric_enum! {
 
 impl Stream {
     /// Create a stream from a VMO
-    pub fn create(
+    pub fn create(vmo: Arc<VmObject>, seek: usize, options: u32) -> Arc<Self> {
-        options: u32,
+        Arc::new(Stream {
-        vmo: Arc<VmObject>,
+            base: KObjectBase::default(),
-        vmo_rights: Rights,
-        seek: usize,
-    ) -> ZxResult<(Arc<Self>, Rights)> {
-        let options = StreamOptions::from_bits(options).ok_or(ZxError::INVALID_ARGS)?;
-        if !(options - StreamOptions::STREAM_CREATE_MASK).is_empty() {
-            return Err(ZxError::INVALID_ARGS);
-        }
-        let mut rights = Rights::DEFAULT_STREAM;
-        if options.contains(StreamOptions::STREAM_MODE_READ) {
-            rights |= Rights::READ;
-            if !vmo_rights.contains(Rights::READ) {
-                return Err(ZxError::ACCESS_DENIED);
-            }
-        }
-        if options.contains(StreamOptions::STREAM_MODE_WRITE) {
-            rights |= Rights::WRITE;
-            if !vmo_rights.contains(Rights::WRITE) {
-                return Err(ZxError::ACCESS_DENIED);
-            }
-        }
-        let out = Arc::new(Stream {
-            base: KObjectBase::with_signal(Signal::empty()), // it seems that stream don't care signals
             options,
             vmo,
             seek: Mutex::new(seek),
-        });
+        })
-        Ok((out, rights))
     }
 
     /// Read data from the stream at the current seek offset
-    pub fn read(&self, data: &OutIoVec<u8>) -> ZxResult<usize> {
+    pub fn read(&self, data: &mut [u8]) -> ZxResult<usize> {
         let mut seek = self.seek.lock();
         let length = self.read_at(data, *seek)?;
         *seek += length;
@@ -85,20 +45,19 @@ impl Stream {
     }
 
     /// Read data from the stream at a given offset
-    pub fn read_at(&self, data: &OutIoVec<u8>, offset: usize) -> ZxResult<usize> {
+    pub fn read_at(&self, data: &mut [u8], offset: usize) -> ZxResult<usize> {
         let count = data.len();
         let content_size = self.vmo.content_size();
         if offset >= content_size {
             return Ok(0);
         }
         let length = count.min(content_size - offset);
-        let slice = data.as_mut_slice()?;
+        self.vmo.read(offset, &mut data[..length])?;
-        self.vmo.read(offset, &mut slice[..length])?;
         Ok(length)
     }
 
     /// write data to the stream at the current seek offset or append data at the end of content
-    pub fn write(&self, data: &InIoVec<u8>, append: bool) -> ZxResult<usize> {
+    pub fn write(&self, data: &[u8], append: bool) -> ZxResult<usize> {
         let mut seek = self.seek.lock();
         if append {
             *seek = self.vmo.content_size();
@@ -109,7 +68,7 @@ impl Stream {
     }
 
     /// Write data to the stream at a given offset
-    pub fn write_at(&self, data: &InIoVec<u8>, offset: usize) -> ZxResult<usize> {
+    pub fn write_at(&self, data: &[u8], offset: usize) -> ZxResult<usize> {
         let count = data.len();
         let mut content_size = self.vmo.content_size();
         let (target_size, overflow) = offset.overflowing_add(count);
@@ -123,8 +82,7 @@ impl Stream {
             return Err(ZxError::NO_SPACE);
         }
         let length = count.min(content_size - offset);
-        let slice = data.as_slice()?;
+        self.vmo.write(offset, &data[..length])?;
-        self.vmo.write(offset, &slice[..length])?;
         Ok(length)
     }
 
@@ -156,7 +114,7 @@ impl Stream {
     pub fn get_info(&self) -> StreamInfo {
         let seek = self.seek.lock();
         StreamInfo {
-            options: self.options.bits,
+            options: self.options,
             padding1: 0,
             seek: *seek as u64,
             content_size: self.vmo.content_size() as u64,
@@ -167,20 +125,20 @@ impl Stream {
 #[repr(C)]
 #[derive(Default)]
 pub struct StreamInfo {
-    // The options passed to zx_stream_create().
+    /// The options passed to `Stream::create()`.
     options: u32,
     padding1: u32,
-    // The current seek offset.
+    /// The current seek offset.
-    //
+    ///
-    // Used by zx_stream_readv and zx_stream_writev to determine where to read
+    /// Used by stream_readv and stream_writev to determine where to read
-    // and write the stream.
+    /// and write the stream.
     seek: u64,
-    // The current size of the stream.
+    /// The current size of the stream.
-    //
+    ///
-    // The number of bytes in the stream that store data. The stream itself
+    /// The number of bytes in the stream that store data. The stream itself
-    // might have a larger capacity to avoid reallocating the underlying storage
+    /// might have a larger capacity to avoid reallocating the underlying storage
-    // as the stream grows or shrinks.
+    /// as the stream grows or shrinks.
-    // NOTE: in fact, this value is store in the VmObject associated and can be
+    /// NOTE: in fact, this value is store in the VmObject associated and can be
-    // get/set through 'object_[get/set]_property(vmo_handle, ...)'
+    /// get/set through 'object_[get/set]_property(vmo_handle, ...)'
     content_size: u64,
 }

zircon-syscall/src/lib.rs

@@ -17,7 +17,7 @@ use {
         sync::atomic::{AtomicI32, Ordering},
     },
     futures::pin_mut,
-    kernel_hal::{user::*, user_io_vec::*, GeneralRegs},
+    kernel_hal::{user::*, GeneralRegs},
     zircon_object::object::*,
     zircon_object::task::Thread,
 };

zircon-syscall/src/stream.rs

@@ -1,4 +1,4 @@
-use {super::*, zircon_object::vm::*};
+use {super::*, bitflags::bitflags, zircon_object::vm::*};
 
 impl Syscall<'_> {
     pub fn sys_stream_create(
@@ -9,13 +9,30 @@ impl Syscall<'_> {
         mut out: UserOutPtr<HandleValue>,
     ) -> ZxResult {
         info!(
-            "stream.create: options={:#x?}, vmo_handle = {:#x?}, seek = {:#x?}",
+            "stream.create: options={:#x?}, vmo_handle={:#x?}, seek={:#x?}",
             options, vmo_handle, seek
         );
+        bitflags! {
+            struct CreateOptions: u32 {
+                #[allow(clippy::identity_op)]
+                const MODE_READ     = 1 << 0;
+                const MODE_WRITE    = 1 << 1;
+            }
+        }
+        let options = CreateOptions::from_bits(options).ok_or(ZxError::INVALID_ARGS)?;
+        let mut rights = Rights::DEFAULT_STREAM;
+        let mut vmo_rights = Rights::empty();
+        if options.contains(CreateOptions::MODE_READ) {
+            rights |= Rights::READ;
+            vmo_rights |= Rights::READ;
+        }
+        if options.contains(CreateOptions::MODE_WRITE) {
+            rights |= Rights::WRITE;
+            vmo_rights |= Rights::WRITE;
+        }
         let proc = self.thread.proc();
-        let (vmo, vmo_rights) = proc.get_object_and_rights::<VmObject>(vmo_handle)?;
+        let vmo = proc.get_object_with_rights::<VmObject>(vmo_handle, vmo_rights)?;
-        let (stream, rights) = Stream::create(options, vmo, vmo_rights, seek)?;
+        let stream = Stream::create(vmo, seek, options.bits());
-        let proc = self.thread.proc();
         let handle = proc.add_handle(Handle::new(stream, rights));
         out.write(handle)?;
         Ok(())
@@ -25,7 +42,7 @@ impl Syscall<'_> {
         &self,
         handle_value: HandleValue,
         options: u32,
-        user_bytes: UserInPtr<InIoVec<u8>>,
+        user_bytes: UserInPtr<IoVecIn>,
         count: usize,
         mut actual_count_ptr: UserOutPtr<usize>,
     ) -> ZxResult {
@@ -33,23 +50,19 @@ impl Syscall<'_> {
             "stream.write: stream={:#x?}, options={:#x?}, buffer={:#x?}, size={:#x?}",
             handle_value, options, user_bytes, count,
         );
-        let options = StreamOptions::from_bits(options).ok_or(ZxError::INVALID_ARGS)?;
+        bitflags! {
-        if !(options - StreamOptions::STREAM_APPEND).is_empty() {
+            struct WriteOptions: u32 {
-            return Err(ZxError::INVALID_ARGS);
+                const APPEND = 1;
             }
-        if user_bytes.is_null() {
-            return Err(ZxError::INVALID_ARGS);
         }
+        let data = user_bytes.read_iovecs(count)?;
+        let options = WriteOptions::from_bits(options).ok_or(ZxError::INVALID_ARGS)?;
         let proc = self.thread.proc();
         let stream = proc.get_object_with_rights::<Stream>(handle_value, Rights::WRITE)?;
-        let data = user_bytes.read_array(count)?;
-        check_total_capacity(&data)?;
         let mut actual_count = 0;
-        for io_vec in &data {
+        for io_vec in data.iter() {
-            if io_vec.is_null() {
+            actual_count +=
-                return Err(ZxError::NOT_FOUND);
+                stream.write(io_vec.as_slice()?, options.contains(WriteOptions::APPEND))?;
-            }
-            actual_count += stream.write(io_vec, options.contains(StreamOptions::STREAM_APPEND))?;
         }
         actual_count_ptr.write_if_not_null(actual_count)?;
         Ok(())
@@ -60,31 +73,24 @@ impl Syscall<'_> {
         handle_value: HandleValue,
         options: u32,
         offset: usize,
-        user_bytes: UserInPtr<InIoVec<u8>>,
+        user_bytes: UserInPtr<IoVecIn>,
         count: usize,
         mut actual_count_ptr: UserOutPtr<usize>,
     ) -> ZxResult {
         info!(
-            "stream.write_at: stream={:#x?}, options={:#x?}, offset = {:#x?}, buffer={:#x?}, size={:#x?}",
+            "stream.write_at: stream={:#x?}, options={:#x?}, offset={:#x?}, buffer={:#x?}, size={:#x?}",
             handle_value, options, offset, user_bytes, count,
         );
         if options != 0 {
             return Err(ZxError::INVALID_ARGS);
         }
-        if user_bytes.is_null() {
+        let data = user_bytes.read_iovecs(count)?;
-            return Err(ZxError::INVALID_ARGS);
-        }
         let proc = self.thread.proc();
         let stream = proc.get_object_with_rights::<Stream>(handle_value, Rights::WRITE)?;
-        let data = user_bytes.read_array(count)?;
-        check_total_capacity(&data)?;
         let mut actual_count = 0;
         let mut off = offset;
-        for io_vec in &data {
+        for io_vec in data.iter() {
-            if io_vec.is_null() {
+            actual_count += stream.write_at(io_vec.as_slice()?, off)?;
-                return Err(ZxError::NOT_FOUND);
-            }
-            actual_count += stream.write_at(io_vec, off)?;
             off += actual_count;
         }
         actual_count_ptr.write_if_not_null(actual_count)?;
@@ -95,7 +101,7 @@ impl Syscall<'_> {
         &self,
         handle_value: HandleValue,
         options: u32,
-        user_bytes: UserInOutPtr<OutIoVec<u8>>,
+        user_bytes: UserInPtr<IoVecOut>,
         count: usize,
         mut actual_count_ptr: UserOutPtr<usize>,
     ) -> ZxResult {
@@ -106,19 +112,12 @@ impl Syscall<'_> {
         if options != 0 {
             return Err(ZxError::INVALID_ARGS);
         }
-        if user_bytes.is_null() {
+        let mut data = user_bytes.read_iovecs(count)?;
-            return Err(ZxError::INVALID_ARGS);
-        }
-        let mut data = user_bytes.read_array(count)?;
         let proc = self.thread.proc();
         let stream = proc.get_object_with_rights::<Stream>(handle_value, Rights::READ)?;
-        check_total_capacity(&data)?;
         let mut actual_count = 0usize;
-        for io_vec in &mut data {
+        for io_vec in data.iter_mut() {
-            if io_vec.is_null() {
+            actual_count += stream.read(io_vec.as_mut_slice()?)?;
-                return Err(ZxError::NOT_FOUND);
-            }
-            actual_count += stream.read(io_vec)?;
         }
         actual_count_ptr.write_if_not_null(actual_count)?;
         Ok(())
@@ -129,31 +128,24 @@ impl Syscall<'_> {
         handle_value: HandleValue,
         options: u32,
         offset: usize,
-        user_bytes: UserInOutPtr<OutIoVec<u8>>,
+        user_bytes: UserInPtr<IoVecOut>,
         count: usize,
         mut actual_count_ptr: UserOutPtr<usize>,
     ) -> ZxResult {
         info!(
-            "stream.read_at: stream={:#x?}, options={:#x?}, offset = {:#x?}, buffer={:#x?}, size={:#x?}",
+            "stream.read_at: stream={:#x?}, options={:#x?}, offset={:#x?}, buffer={:#x?}, size={:#x?}",
             handle_value, options, offset, user_bytes, count,
         );
         if options != 0 {
             return Err(ZxError::INVALID_ARGS);
         }
-        if user_bytes.is_null() {
+        let mut data = user_bytes.read_iovecs(count)?;
-            return Err(ZxError::INVALID_ARGS);
-        }
-        let mut data = user_bytes.read_array(count)?;
         let proc = self.thread.proc();
         let stream = proc.get_object_with_rights::<Stream>(handle_value, Rights::READ)?;
-        check_total_capacity(&data)?;
         let mut actual_count = 0usize;
         let mut off = offset;
-        for io_vec in &mut data {
+        for io_vec in data.iter_mut() {
-            if io_vec.is_null() {
+            actual_count += stream.read_at(io_vec.as_mut_slice()?, off)?;
-                return Err(ZxError::NOT_FOUND);
-            }
-            actual_count += stream.read_at(io_vec, off)?;
             off += actual_count;
         }
         actual_count_ptr.write_if_not_null(actual_count)?;
@@ -168,7 +160,7 @@ impl Syscall<'_> {
         mut out_seek: UserOutPtr<usize>,
     ) -> ZxResult {
         info!(
-            "stream.seek: stream={:#x?}, seek_origin={:#x?}, offset = {:#x?}",
+            "stream.seek: stream={:#x?}, seek_origin={:#x?}, offset={:#x?}",
             handle_value, seek_origin, offset,
         );
         let proc = self.thread.proc();
@@ -182,15 +174,3 @@ impl Syscall<'_> {
         Ok(())
     }
 }
-
-fn check_total_capacity<T, P: Policy>(data: &[IoVec<T, P>]) -> ZxResult {
-    let mut total_count = 0usize;
-    for io_vec in data {
-        let (result, overflow) = total_count.overflowing_add(io_vec.len());
-        if overflow {
-            return Err(ZxError::INVALID_ARGS);
-        }
-        total_count = result;
-    }
-    Ok(())
-}