bjoernager/rust
bjoernager/rust
1
Fork 0
Code Activity

Merge pull request #4174 from RalfJung/read-write-callback

Browse source 
files: make read/write take callback to store result
This commit is contained in:
Ralf Jung 2025-02-02 19:13:49 +00:00 • committed by GitHub
commit 6b656cc0c7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 158 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

97
src/tools/miri/src/shims/files.rs
View file

@ -1,6 +1,6 @@
use std::any::Any;
use std::collections::BTreeMap;
use std::io::{IsTerminal, Read, SeekFrom, Write};
use std::io::{IsTerminal, SeekFrom, Write};
use std::marker::CoercePointee;
use std::ops::Deref;
use std::rc::{Rc, Weak};
@ -140,8 +140,8 @@ pub trait FileDescription: std::fmt::Debug + FileDescriptionExt {
_communicate_allowed: bool,
_ptr: Pointer,
_len: usize,
_dest: &MPlaceTy<'tcx>,
_ecx: &mut MiriInterpCx<'tcx>,
_finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
throw_unsup_format!("cannot read from {}", self.name());
}
@ -154,8 +154,8 @@ pub trait FileDescription: std::fmt::Debug + FileDescriptionExt {
_communicate_allowed: bool,
_ptr: Pointer,
_len: usize,
_dest: &MPlaceTy<'tcx>,
_ecx: &mut MiriInterpCx<'tcx>,
_finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
throw_unsup_format!("cannot write to {}", self.name());
}
@ -207,19 +207,16 @@ impl FileDescription for io::Stdin {
communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
let mut bytes = vec![0; len];
if !communicate_allowed {
// We want isolation mode to be deterministic, so we have to disallow all reads, even stdin.
helpers::isolation_abort_error("`read` from stdin")?;
}
let result = Read::read(&mut &*self, &mut bytes);
match result {
Ok(read_size) => ecx.return_read_success(ptr, &bytes, read_size, dest),
Err(e) => ecx.set_last_error_and_return(e, dest),
}
let result = ecx.read_from_host(&*self, len, ptr)?;
finish.call(ecx, result)
}
fn is_tty(&self, communicate_allowed: bool) -> bool {
@ -237,22 +234,19 @@ impl FileDescription for io::Stdout {
_communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
let bytes = ecx.read_bytes_ptr_strip_provenance(ptr, Size::from_bytes(len))?;
// We allow writing to stderr even with isolation enabled.
let result = Write::write(&mut &*self, bytes);
// We allow writing to stdout even with isolation enabled.
let result = ecx.write_to_host(&*self, len, ptr)?;
// Stdout is buffered, flush to make sure it appears on the
// screen. This is the write() syscall of the interpreted
// program, we want it to correspond to a write() syscall on
// the host -- there is no good in adding extra buffering
// here.
io::stdout().flush().unwrap();
match result {
Ok(write_size) => ecx.return_write_success(write_size, dest),
Err(e) => ecx.set_last_error_and_return(e, dest),
}
finish.call(ecx, result)
}
fn is_tty(&self, communicate_allowed: bool) -> bool {
@ -270,17 +264,13 @@ impl FileDescription for io::Stderr {
_communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
let bytes = ecx.read_bytes_ptr_strip_provenance(ptr, Size::from_bytes(len))?;
// We allow writing to stderr even with isolation enabled.
let result = ecx.write_to_host(&*self, len, ptr)?;
// No need to flush, stderr is not buffered.
let result = Write::write(&mut &*self, bytes);
match result {
Ok(write_size) => ecx.return_write_success(write_size, dest),
Err(e) => ecx.set_last_error_and_return(e, dest),
}
finish.call(ecx, result)
}
fn is_tty(&self, communicate_allowed: bool) -> bool {
@ -302,11 +292,11 @@ impl FileDescription for NullOutput {
_communicate_allowed: bool,
_ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
// We just don't write anything, but report to the user that we did.
ecx.return_write_success(len, dest)
finish.call(ecx, Ok(len))
}
}
@ -405,40 +395,41 @@ impl FdTable {
impl<'tcx> EvalContextExt<'tcx> for crate::MiriInterpCx<'tcx> {}
pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
/// Helper to implement `FileDescription::read`:
/// This is only used when `read` is successful.
/// `actual_read_size` should be the return value of some underlying `read` call that used
/// `bytes` as its output buffer.
/// The length of `bytes` must not exceed either the host's or the target's `isize`.
/// `bytes` is written to `buf` and the size is written to `dest`.
fn return_read_success(
/// Read data from a host `Read` type, store the result into machine memory,
/// and return whether that worked.
fn read_from_host(
&mut self,
buf: Pointer,
bytes: &[u8],
actual_read_size: usize,
dest: &MPlaceTy<'tcx>,
) -> InterpResult<'tcx> {
mut file: impl io::Read,
len: usize,
ptr: Pointer,
) -> InterpResult<'tcx, Result<usize, IoError>> {
let this = self.eval_context_mut();
let mut bytes = vec![0; len];
let result = file.read(&mut bytes);
match result {
Ok(read_size) => {
// If reading to `bytes` did not fail, we write those bytes to the buffer.
// Crucially, if fewer than `bytes.len()` bytes were read, only write
// that much into the output buffer!
this.write_bytes_ptr(buf, bytes[..actual_read_size].iter().copied())?;
// The actual read size is always less than what got originally requested so this cannot fail.
this.write_int(u64::try_from(actual_read_size).unwrap(), dest)?;
interp_ok(())
this.write_bytes_ptr(ptr, bytes[..read_size].iter().copied())?;
interp_ok(Ok(read_size))
}
Err(e) => interp_ok(Err(IoError::HostError(e))),
}
}
/// Helper to implement `FileDescription::write`:
/// This function is only used when `write` is successful, and writes `actual_write_size` to `dest`
fn return_write_success(
/// Write data to a host `Write` type, withthe bytes taken from machine memory.
fn write_to_host(
&mut self,
actual_write_size: usize,
dest: &MPlaceTy<'tcx>,
) -> InterpResult<'tcx> {
mut file: impl io::Write,
len: usize,
ptr: Pointer,
) -> InterpResult<'tcx, Result<usize, IoError>> {
let this = self.eval_context_mut();
// The actual write size is always less than what got originally requested so this cannot fail.
this.write_int(u64::try_from(actual_write_size).unwrap(), dest)?;
interp_ok(())
let bytes = this.read_bytes_ptr_strip_provenance(ptr, Size::from_bytes(len))?;
let result = file.write(bytes);
interp_ok(result.map_err(IoError::HostError))
}
}

54
src/tools/miri/src/shims/unix/fd.rs
View file

@ -30,8 +30,8 @@ pub trait UnixFileDescription: FileDescription {
_offset: u64,
_ptr: Pointer,
_len: usize,
_dest: &MPlaceTy<'tcx>,
_ecx: &mut MiriInterpCx<'tcx>,
_finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
throw_unsup_format!("cannot pread from {}", self.name());
}
@ -46,8 +46,8 @@ pub trait UnixFileDescription: FileDescription {
_ptr: Pointer,
_len: usize,
_offset: u64,
_dest: &MPlaceTy<'tcx>,
_ecx: &mut MiriInterpCx<'tcx>,
_finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
throw_unsup_format!("cannot pwrite to {}", self.name());
}
@ -236,7 +236,7 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
let count = usize::try_from(count).unwrap(); // now it fits in a `usize`
let communicate = this.machine.communicate();
// We temporarily dup the FD to be able to retain mutable access to `this`.
// Get the FD.
let Some(fd) = this.machine.fds.get(fd_num) else {
trace!("read: FD not found");
return this.set_last_error_and_return(LibcError("EBADF"), dest);
@ -247,13 +247,33 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
// because it was a target's `usize`. Also we are sure that its smaller than
// `usize::MAX` because it is bounded by the host's `isize`.
let finish = {
let dest = dest.clone();
callback!(
@capture<'tcx> {
count: usize,
dest: MPlaceTy<'tcx>,
}
|this, result: Result<usize, IoError>| {
match result {
Ok(read_size) => {
assert!(read_size <= count);
// This must fit since `count` fits.
this.write_int(u64::try_from(read_size).unwrap(), &dest)
}
Err(e) => {
this.set_last_error_and_return(e, &dest)
}
}}
)
};
match offset {
None => fd.read(communicate, buf, count, dest, this)?,
None => fd.read(communicate, buf, count, this, finish)?,
Some(offset) => {
let Ok(offset) = u64::try_from(offset) else {
return this.set_last_error_and_return(LibcError("EINVAL"), dest);
};
fd.as_unix().pread(communicate, offset, buf, count, dest, this)?
fd.as_unix().pread(communicate, offset, buf, count, this, finish)?
}
};
interp_ok(())
@ -287,13 +307,33 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
return this.set_last_error_and_return(LibcError("EBADF"), dest);
};
let finish = {
let dest = dest.clone();
callback!(
@capture<'tcx> {
count: usize,
dest: MPlaceTy<'tcx>,
}
|this, result: Result<usize, IoError>| {
match result {
Ok(write_size) => {
assert!(write_size <= count);
// This must fit since `count` fits.
this.write_int(u64::try_from(write_size).unwrap(), &dest)
}
Err(e) => {
this.set_last_error_and_return(e, &dest)
}
}}
)
};
match offset {
None => fd.write(communicate, buf, count, dest, this)?,
None => fd.write(communicate, buf, count, this, finish)?,
Some(offset) => {
let Ok(offset) = u64::try_from(offset) else {
return this.set_last_error_and_return(LibcError("EINVAL"), dest);
};
fd.as_unix().pwrite(communicate, buf, count, offset, dest, this)?
fd.as_unix().pwrite(communicate, buf, count, offset, this, finish)?
}
};
interp_ok(())

45
src/tools/miri/src/shims/unix/fs.rs
View file

@ -35,16 +35,13 @@ impl FileDescription for FileHandle {
communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
assert!(communicate_allowed, "isolation should have prevented even opening a file");
let mut bytes = vec![0; len];
let result = (&mut &self.file).read(&mut bytes);
match result {
Ok(read_size) => ecx.return_read_success(ptr, &bytes, read_size, dest),
Err(e) => ecx.set_last_error_and_return(e, dest),
}
let result = ecx.read_from_host(&self.file, len, ptr)?;
finish.call(ecx, result)
}
fn write<'tcx>(
@ -52,16 +49,13 @@ impl FileDescription for FileHandle {
communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
assert!(communicate_allowed, "isolation should have prevented even opening a file");
let bytes = ecx.read_bytes_ptr_strip_provenance(ptr, Size::from_bytes(len))?;
let result = (&mut &self.file).write(bytes);
match result {
Ok(write_size) => ecx.return_write_success(write_size, dest),
Err(e) => ecx.set_last_error_and_return(e, dest),
}
let result = ecx.write_to_host(&self.file, len, ptr)?;
finish.call(ecx, result)
}
fn seek<'tcx>(
@ -119,8 +113,8 @@ impl UnixFileDescription for FileHandle {
offset: u64,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
assert!(communicate_allowed, "isolation should have prevented even opening a file");
let mut bytes = vec![0; len];
@ -137,11 +131,17 @@ impl UnixFileDescription for FileHandle {
.expect("failed to restore file position, this shouldn't be possible");
res
};
let result = f();
match result {
Ok(read_size) => ecx.return_read_success(ptr, &bytes, read_size, dest),
Err(e) => ecx.set_last_error_and_return(e, dest),
let result = match f() {
Ok(read_size) => {
// If reading to `bytes` did not fail, we write those bytes to the buffer.
// Crucially, if fewer than `bytes.len()` bytes were read, only write
// that much into the output buffer!
ecx.write_bytes_ptr(ptr, bytes[..read_size].iter().copied())?;
Ok(read_size)
}
Err(e) => Err(IoError::HostError(e)),
};
finish.call(ecx, result)
}
fn pwrite<'tcx>(
@ -150,8 +150,8 @@ impl UnixFileDescription for FileHandle {
ptr: Pointer,
len: usize,
offset: u64,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
assert!(communicate_allowed, "isolation should have prevented even opening a file");
// Emulates pwrite using seek + write + seek to restore cursor position.
@ -169,10 +169,7 @@ impl UnixFileDescription for FileHandle {
res
};
let result = f();
match result {
Ok(write_size) => ecx.return_write_success(write_size, dest),
Err(e) => ecx.set_last_error_and_return(e, dest),
}
finish.call(ecx, result.map_err(IoError::HostError))
}
fn flock<'tcx>(

37
src/tools/miri/src/shims/unix/linux_like/eventfd.rs
View file

@ -51,20 +51,20 @@ impl FileDescription for EventFd {
_communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
// We're treating the buffer as a `u64`.
let ty = ecx.machine.layouts.u64;
// Check the size of slice, and return error only if the size of the slice < 8.
if len < ty.size.bytes_usize() {
return ecx.set_last_error_and_return(ErrorKind::InvalidInput, dest);
return finish.call(ecx, Err(ErrorKind::InvalidInput.into()));
}
// Turn the pointer into a place at the right type.
let buf_place = ecx.ptr_to_mplace_unaligned(ptr, ty);
eventfd_read(buf_place, dest, self, ecx)
eventfd_read(buf_place, self, ecx, finish)
}
/// A write call adds the 8-byte integer value supplied in
@ -84,20 +84,20 @@ impl FileDescription for EventFd {
_communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
// We're treating the buffer as a `u64`.
let ty = ecx.machine.layouts.u64;
// Check the size of slice, and return error only if the size of the slice < 8.
if len < ty.layout.size.bytes_usize() {
return ecx.set_last_error_and_return(ErrorKind::InvalidInput, dest);
return finish.call(ecx, Err(ErrorKind::InvalidInput.into()));
}
// Turn the pointer into a place at the right type.
let buf_place = ecx.ptr_to_mplace_unaligned(ptr, ty);
eventfd_write(buf_place, dest, self, ecx)
eventfd_write(buf_place, self, ecx, finish)
}
fn as_unix(&self) -> &dyn UnixFileDescription {
@ -183,15 +183,15 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
/// else just add the user-supplied value to current counter.
fn eventfd_write<'tcx>(
buf_place: MPlaceTy<'tcx>,
dest: &MPlaceTy<'tcx>,
eventfd: FileDescriptionRef<EventFd>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
// Figure out which value we should add.
let num = ecx.read_scalar(&buf_place)?.to_u64()?;
// u64::MAX as input is invalid because the maximum value of counter is u64::MAX - 1.
if num == u64::MAX {
return ecx.set_last_error_and_return(ErrorKind::InvalidInput, dest);
return finish.call(ecx, Err(ErrorKind::InvalidInput.into()));
}
match eventfd.counter.get().checked_add(num) {
@ -219,16 +219,14 @@ fn eventfd_write<'tcx>(
ecx.check_and_update_readiness(eventfd)?;
// Return how many bytes we consumed from the user-provided buffer.
return ecx.write_int(buf_place.layout.size.bytes(), dest);
return finish.call(ecx, Ok(buf_place.layout.size.bytes_usize()));
}
None | Some(u64::MAX) => {
// We can't update the state, so we have to block.
if eventfd.is_nonblock {
return ecx.set_last_error_and_return(ErrorKind::WouldBlock, dest);
return finish.call(ecx, Err(ErrorKind::WouldBlock.into()));
}
let dest = dest.clone();
eventfd.blocked_write_tid.borrow_mut().push(ecx.active_thread());
let weak_eventfd = FileDescriptionRef::downgrade(&eventfd);
@ -239,7 +237,7 @@ fn eventfd_write<'tcx>(
@capture<'tcx> {
num: u64,
buf_place: MPlaceTy<'tcx>,
dest: MPlaceTy<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
weak_eventfd: WeakFileDescriptionRef<EventFd>,
}
|this, unblock: UnblockKind| {
@ -247,7 +245,7 @@ fn eventfd_write<'tcx>(
// When we get unblocked, try again. We know the ref is still valid,
// otherwise there couldn't be a `write` that unblocks us.
let eventfd_ref = weak_eventfd.upgrade().unwrap();
eventfd_write(buf_place, &dest, eventfd_ref, this)
eventfd_write(buf_place, eventfd_ref, this, finish)
}
),
);
@ -260,9 +258,9 @@ fn eventfd_write<'tcx>(
/// else just return the current counter value to the caller and set the counter to 0.
fn eventfd_read<'tcx>(
buf_place: MPlaceTy<'tcx>,
dest: &MPlaceTy<'tcx>,
eventfd: FileDescriptionRef<EventFd>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
// Set counter to 0, get old value.
let counter = eventfd.counter.replace(0);
@ -270,9 +268,8 @@ fn eventfd_read<'tcx>(
// Block when counter == 0.
if counter == 0 {
if eventfd.is_nonblock {
return ecx.set_last_error_and_return(ErrorKind::WouldBlock, dest);
return finish.call(ecx, Err(ErrorKind::WouldBlock.into()));
}
let dest = dest.clone();
eventfd.blocked_read_tid.borrow_mut().push(ecx.active_thread());
@ -283,7 +280,7 @@ fn eventfd_read<'tcx>(
callback!(
@capture<'tcx> {
buf_place: MPlaceTy<'tcx>,
dest: MPlaceTy<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
weak_eventfd: WeakFileDescriptionRef<EventFd>,
}
|this, unblock: UnblockKind| {
@ -291,7 +288,7 @@ fn eventfd_read<'tcx>(
// When we get unblocked, try again. We know the ref is still valid,
// otherwise there couldn't be a `write` that unblocks us.
let eventfd_ref = weak_eventfd.upgrade().unwrap();
eventfd_read(buf_place, &dest, eventfd_ref, this)
eventfd_read(buf_place, eventfd_ref, this, finish)
}
),
);
@ -317,7 +314,7 @@ fn eventfd_read<'tcx>(
ecx.check_and_update_readiness(eventfd)?;
// Tell userspace how many bytes we put into the buffer.
return ecx.write_int(buf_place.layout.size.bytes(), dest);
return finish.call(ecx, Ok(buf_place.layout.size.bytes_usize()));
}
interp_ok(())
}

57
src/tools/miri/src/shims/unix/unnamed_socket.rs
View file

@ -5,9 +5,7 @@
use std::cell::{Cell, OnceCell, RefCell};
use std::collections::VecDeque;
use std::io;
use std::io::{ErrorKind, Read};
use rustc_abi::Size;
use std::io::ErrorKind;
use crate::concurrency::VClock;
use crate::shims::files::{
@ -92,10 +90,10 @@ impl FileDescription for AnonSocket {
_communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
anonsocket_read(self, len, ptr, dest, ecx)
anonsocket_read(self, ptr, len, ecx, finish)
}
fn write<'tcx>(
@ -103,10 +101,10 @@ impl FileDescription for AnonSocket {
_communicate_allowed: bool,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
anonsocket_write(self, ptr, len, dest, ecx)
anonsocket_write(self, ptr, len, ecx, finish)
}
fn as_unix(&self) -> &dyn UnixFileDescription {
@ -119,25 +117,25 @@ fn anonsocket_write<'tcx>(
self_ref: FileDescriptionRef<AnonSocket>,
ptr: Pointer,
len: usize,
dest: &MPlaceTy<'tcx>,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
// Always succeed on write size 0.
// ("If count is zero and fd refers to a file other than a regular file, the results are not specified.")
if len == 0 {
return ecx.return_write_success(0, dest);
return finish.call(ecx, Ok(0));
}
// We are writing to our peer's readbuf.
let Some(peer_fd) = self_ref.peer_fd().upgrade() else {
// If the upgrade from Weak to Rc fails, it indicates that all read ends have been
// closed. It is an error to write even if there would be space.
return ecx.set_last_error_and_return(ErrorKind::BrokenPipe, dest);
return finish.call(ecx, Err(ErrorKind::BrokenPipe.into()));
};
let Some(writebuf) = &peer_fd.readbuf else {
// Writing to the read end of a pipe.
return ecx.set_last_error_and_return(IoError::LibcError("EBADF"), dest);
return finish.call(ecx, Err(IoError::LibcError("EBADF")));
};
// Let's see if we can write.
@ -145,13 +143,12 @@ fn anonsocket_write<'tcx>(
if available_space == 0 {
if self_ref.is_nonblock {
// Non-blocking socketpair with a full buffer.
return ecx.set_last_error_and_return(ErrorKind::WouldBlock, dest);
return finish.call(ecx, Err(ErrorKind::WouldBlock.into()));
} else {
self_ref.blocked_write_tid.borrow_mut().push(ecx.active_thread());
// Blocking socketpair with a full buffer.
// Block the current thread; only keep a weak ref for this.
let weak_self_ref = FileDescriptionRef::downgrade(&self_ref);
let dest = dest.clone();
ecx.block_thread(
BlockReason::UnnamedSocket,
None,
@ -160,14 +157,14 @@ fn anonsocket_write<'tcx>(
weak_self_ref: WeakFileDescriptionRef<AnonSocket>,
ptr: Pointer,
len: usize,
dest: MPlaceTy<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
}
|this, unblock: UnblockKind| {
assert_eq!(unblock, UnblockKind::Ready);
// If we got unblocked, then our peer successfully upgraded its weak
// ref to us. That means we can also upgrade our weak ref.
let self_ref = weak_self_ref.upgrade().unwrap();
anonsocket_write(self_ref, ptr, len, &dest, this)
anonsocket_write(self_ref, ptr, len, this, finish)
}
),
);
@ -180,9 +177,9 @@ fn anonsocket_write<'tcx>(
writebuf.clock.join(clock);
});
// Do full write / partial write based on the space available.
let actual_write_size = len.min(available_space);
let bytes = ecx.read_bytes_ptr_strip_provenance(ptr, Size::from_bytes(len))?;
writebuf.buf.extend(&bytes[..actual_write_size]);
let write_size = len.min(available_space);
let actual_write_size = ecx.write_to_host(&mut writebuf.buf, write_size, ptr)?.unwrap();
assert_eq!(actual_write_size, write_size);
// Need to stop accessing peer_fd so that it can be notified.
drop(writebuf);
@ -197,7 +194,7 @@ fn anonsocket_write<'tcx>(
// The kernel does this even if the fd was already readable before, so we follow suit.
ecx.check_and_update_readiness(peer_fd)?;
return ecx.return_write_success(actual_write_size, dest);
return finish.call(ecx, Ok(write_size));
}
interp_ok(())
}
@ -205,14 +202,14 @@ fn anonsocket_write<'tcx>(
/// Read from AnonSocket and return the number of bytes read.
fn anonsocket_read<'tcx>(
self_ref: FileDescriptionRef<AnonSocket>,
len: usize,
ptr: Pointer,
dest: &MPlaceTy<'tcx>,
len: usize,
ecx: &mut MiriInterpCx<'tcx>,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
) -> InterpResult<'tcx> {
// Always succeed on read size 0.
if len == 0 {
return ecx.return_read_success(ptr, &[], 0, dest);
return finish.call(ecx, Ok(0));
}
let Some(readbuf) = &self_ref.readbuf else {
@ -225,43 +222,41 @@ fn anonsocket_read<'tcx>(
if self_ref.peer_fd().upgrade().is_none() {
// Socketpair with no peer and empty buffer.
// 0 bytes successfully read indicates end-of-file.
return ecx.return_read_success(ptr, &[], 0, dest);
return finish.call(ecx, Ok(0));
} else if self_ref.is_nonblock {
// Non-blocking socketpair with writer and empty buffer.
// https://linux.die.net/man/2/read
// EAGAIN or EWOULDBLOCK can be returned for socket,
// POSIX.1-2001 allows either error to be returned for this case.
// Since there is no ErrorKind for EAGAIN, WouldBlock is used.
return ecx.set_last_error_and_return(ErrorKind::WouldBlock, dest);
return finish.call(ecx, Err(ErrorKind::WouldBlock.into()));
} else {
self_ref.blocked_read_tid.borrow_mut().push(ecx.active_thread());
// Blocking socketpair with writer and empty buffer.
// Block the current thread; only keep a weak ref for this.
let weak_self_ref = FileDescriptionRef::downgrade(&self_ref);
let dest = dest.clone();
ecx.block_thread(
BlockReason::UnnamedSocket,
None,
callback!(
@capture<'tcx> {
weak_self_ref: WeakFileDescriptionRef<AnonSocket>,
len: usize,
ptr: Pointer,
dest: MPlaceTy<'tcx>,
len: usize,
finish: DynMachineCallback<'tcx, Result<usize, IoError>>,
}
|this, unblock: UnblockKind| {
assert_eq!(unblock, UnblockKind::Ready);
// If we got unblocked, then our peer successfully upgraded its weak
// ref to us. That means we can also upgrade our weak ref.
let self_ref = weak_self_ref.upgrade().unwrap();
anonsocket_read(self_ref, len, ptr, &dest, this)
anonsocket_read(self_ref, ptr, len, this, finish)
}
),
);
}
} else {
// There's data to be read!
let mut bytes = vec![0; len];
let mut readbuf = readbuf.borrow_mut();
// Synchronize with all previous writes to this buffer.
// FIXME: this over-synchronizes; a more precise approach would be to
@ -270,7 +265,7 @@ fn anonsocket_read<'tcx>(
// Do full read / partial read based on the space available.
// Conveniently, `read` exists on `VecDeque` and has exactly the desired behavior.
let actual_read_size = readbuf.buf.read(&mut bytes[..]).unwrap();
let read_size = ecx.read_from_host(&mut readbuf.buf, len, ptr)?.unwrap();
// Need to drop before others can access the readbuf again.
drop(readbuf);
@ -293,7 +288,7 @@ fn anonsocket_read<'tcx>(
ecx.check_and_update_readiness(peer_fd)?;
};
return ecx.return_read_success(ptr, &bytes, actual_read_size, dest);
return finish.call(ecx, Ok(read_size));
}
interp_ok(())
}