Repository: bytecodealliance/wasmtime
Pull request #9232 (2024-09-12 15:18:02)

Add a `Signed` trait to match the `Unsigned` trait
I've wanted to use this every now and then but haven't gotten around to adding this so figured I'd go ahead and add it.

<!--
Please make sure you include the following information:

- If this work has been discussed elsewhere, please include a link to that
  conversation. If it was discussed in an issue, just mention "issue #...".

- Explain why this change is needed. If the details are in an issue already,
  this can be brief.

Our development process is documented in the Wasmtime book:
https://docs.wasmtime.dev/contributing-development-process.html

Please ensure all communication follows the code of conduct:
https://github.com/bytecodealliance/wasmtime/blob/main/CODE_OF_CONDUCT.md
-->



Patch:
diff --git a/.github/actions/install-rust/action.yml b/.github/actions/install-rust/action.yml
index 55ba39879..e0ebe61c4 100644
--- a/.github/actions/install-rust/action.yml
+++ b/.github/actions/install-rust/action.yml
@@ -6,6 +6,18 @@ inputs:
     description: 'Default toolchan to install'
     required: false
     default: 'default'
+  msrv_range:
+    description: 'Versions later-than-latest-Rust the MSRV supports'
+    required: false
+    # Note that this is currently set to 3 as the MSRV is 1.78 and current
+    # stable is 1.81. Currently MSRV cannot be updated as it would break the
+    # build on OSS-Fuzz which hasn't updated its Rust compiler in quite some time.
+    #
+    # Updating OSS-Fuzz is being done in
+    # https://github.com/google/oss-fuzz/pull/12365 but it's taking some time,
+    # so for now this'll get bumped instead of MSRV while we wait for the update
+    # to happen.
+    default: '3'
 
 runs:
   using: composite
@@ -17,9 +29,10 @@ runs:
         # Determine MSRV as N in `1.N.0` by looking at the `rust-version`
         # located in the root `Cargo.toml`.
         msrv=$(grep 'rust-version.*1' Cargo.toml | sed 's/.*\.\([0-9]*\)\..*/\1/')
+        range=${{ inputs.msrv_range }}
 
         if [ "${{ inputs.toolchain }}" = "default" ]; then
-          echo "version=1.$((msrv+2)).0" >> "$GITHUB_OUTPUT"
+          echo "version=1.$((msrv+range)).0" >> "$GITHUB_OUTPUT"
         elif [ "${{ inputs.toolchain }}" = "msrv" ]; then
           echo "version=1.$msrv.0" >> "$GITHUB_OUTPUT"
         elif [ "${{ inputs.toolchain }}" = "wasmtime-ci-pinned-nightly" ]; then
diff --git a/Cargo.toml b/Cargo.toml
index abe433b4e..32eca81f4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -161,7 +161,7 @@ authors = ["The Wasmtime Project Developers"]
 edition = "2021"
 # Wasmtime's current policy is that this number can be no larger than the
 # current stable release of Rust minus 2.
-rust-version = "1.79.0"
+rust-version = "1.78.0"
 
 [workspace.lints.rust]
 # Turn on some lints which are otherwise allow-by-default in rustc.
diff --git a/cranelift/entity/src/lib.rs b/cranelift/entity/src/lib.rs
index 4cbed9d0a..5ad0014c3 100644
--- a/cranelift/entity/src/lib.rs
+++ b/cranelift/entity/src/lib.rs
@@ -265,6 +265,7 @@ mod list;
 mod map;
 mod primary;
 mod set;
+mod signed;
 mod sparse;
 mod unsigned;
 
@@ -275,6 +276,7 @@ pub use self::list::{EntityList, ListPool};
 pub use self::map::SecondaryMap;
 pub use self::primary::PrimaryMap;
 pub use self::set::EntitySet;
+pub use self::signed::Signed;
 pub use self::sparse::{SparseMap, SparseMapValue, SparseSet};
 pub use self::unsigned::Unsigned;
 
diff --git a/cranelift/entity/src/signed.rs b/cranelift/entity/src/signed.rs
new file mode 100644
index 000000000..3c136d590
--- /dev/null
+++ b/cranelift/entity/src/signed.rs
@@ -0,0 +1,40 @@
+/// Helper trait used to add `signed()` methods to primitive unsigned integer
+/// types.
+///
+/// The purpose of this trait is to signal the intent that the sign bit of an
+/// unsigned integer is intended to be discarded and the value is instead
+/// understood to be a "bag of bits" where the conversion to a signed number
+/// is intended to be lossless bit-wise. This can be used for example when
+/// converting an unsigned integer into a signed integer for constrained reasons
+/// outside the scope of the code in question.
+pub trait Signed {
+    /// The signed integer for this type which has the same width.
+    type Signed;
+
+    /// View this unsigned integer as a signed integer of the same width.
+    ///
+    /// All bits are preserved.
+    fn signed(self) -> Self::Signed;
+}
+
+macro_rules! impls {
+    ($($unsigned:ident => $signed:ident)*) => {$(
+        impl Signed for $unsigned {
+            type Signed = $signed;
+
+            #[inline]
+            fn signed(self) -> $signed {
+                self as $signed
+            }
+        }
+    )*}
+}
+
+impls! {
+    u8 => i8
+    u16 => i16
+    u32 => i32
+    u64 => i64
+    u128 => i128
+    usize => isize
+}
diff --git a/crates/wasi/src/filesystem.rs b/crates/wasi/src/filesystem.rs
index a924fa96a..2c656b699 100644
--- a/crates/wasi/src/filesystem.rs
+++ b/crates/wasi/src/filesystem.rs
@@ -361,7 +361,7 @@ impl HostInputStream for FileInputStream {
                 // background tasks. Also, the guest is only blocked if
                 // the stream was dropped mid-read, which we don't expect to
                 // occur frequently.
-                task.abort_wait().await;
+                task.cancel().await;
             }
             _ => {}
         }
@@ -558,7 +558,7 @@ impl HostOutputStream for FileOutputStream {
                 // background tasks. Also, the guest is only blocked if
                 // the stream was dropped mid-write, which we don't expect to
                 // occur frequently.
-                task.abort_wait().await;
+                task.cancel().await;
             }
             _ => {}
         }
diff --git a/crates/wasi/src/host/tcp.rs b/crates/wasi/src/host/tcp.rs
index 2f20cdc25..143ea607e 100644
--- a/crates/wasi/src/host/tcp.rs
+++ b/crates/wasi/src/host/tcp.rs
@@ -298,7 +298,7 @@ where
             ShutdownType::Send => std::net::Shutdown::Write,
             ShutdownType::Both => std::net::Shutdown::Both,
         };
-        Ok(socket.shutdown(how)?)
+        socket.shutdown(how)
     }
 
     fn drop(&mut self, this: Resource<tcp::TcpSocket>) -> Result<(), anyhow::Error> {
diff --git a/crates/wasi/src/pipe.rs b/crates/wasi/src/pipe.rs
index 4b723cc0c..83a389d90 100644
--- a/crates/wasi/src/pipe.rs
+++ b/crates/wasi/src/pipe.rs
@@ -193,7 +193,7 @@ impl HostInputStream for AsyncReadStream {
 
     async fn cancel(&mut self) {
         match self.join_handle.take() {
-            Some(task) => _ = task.abort_wait().await,
+            Some(task) => _ = task.cancel().await,
             None => {}
         }
     }
diff --git a/crates/wasi/src/runtime.rs b/crates/wasi/src/runtime.rs
index dc29fef97..b1fb78396 100644
--- a/crates/wasi/src/runtime.rs
+++ b/crates/wasi/src/runtime.rs
@@ -42,8 +42,9 @@ pub struct AbortOnDropJoinHandle<T>(tokio::task::JoinHandle<T>);
 impl<T> AbortOnDropJoinHandle<T> {
     /// Abort the task and wait for it to finish. Optionally returns the result
     /// of the task if it ran to completion prior to being aborted.
-    pub(crate) async fn abort_wait(mut self) -> Option<T> {
+    pub(crate) async fn cancel(mut self) -> Option<T> {
         self.0.abort();
+
         match (&mut self.0).await {
             Ok(value) => Some(value),
             Err(err) if err.is_cancelled() => None,
diff --git a/crates/wasi/src/tcp.rs b/crates/wasi/src/tcp.rs
index 32feef95a..44ba68416 100644
--- a/crates/wasi/src/tcp.rs
+++ b/crates/wasi/src/tcp.rs
@@ -3,10 +3,10 @@ use crate::host::network;
 use crate::network::SocketAddressFamily;
 use crate::runtime::{with_ambient_tokio_runtime, AbortOnDropJoinHandle};
 use crate::{
-    HostInputStream, HostOutputStream, InputStream, OutputStream, SocketResult, StreamError,
-    Subscribe,
+    HostInputStream, HostOutputStream, InputStream, OutputStream, SocketError, SocketResult,
+    StreamError, Subscribe,
 };
-use anyhow::{Error, Result};
+use anyhow::Result;
 use cap_net_ext::AddressFamily;
 use futures::Future;
 use io_lifetimes::views::SocketlikeView;
@@ -15,10 +15,11 @@ use rustix::io::Errno;
 use rustix::net::sockopt;
 use std::io;
 use std::mem;
-use std::net::SocketAddr;
+use std::net::{Shutdown, SocketAddr};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::Poll;
+use tokio::sync::Mutex;
 
 /// Value taken from rust std library.
 const DEFAULT_BACKLOG: u32 = 128;
@@ -54,7 +55,13 @@ enum TcpState {
     ConnectReady(io::Result<tokio::net::TcpStream>),
 
     /// An outgoing connection has been established.
-    Connected(Arc<tokio::net::TcpStream>),
+    Connected {
+        stream: Arc<tokio::net::TcpStream>,
+
+        // WASI is single threaded, so in practice these Mutexes should never be contended:
+        reader: Arc<Mutex<TcpReader>>,
+        writer: Arc<Mutex<TcpWriter>>,
+    },
 
     Closed,
 }
@@ -72,7 +79,7 @@ impl std::fmt::Debug for TcpState {
                 .finish(),
             Self::Connecting(_) => f.debug_tuple("Connecting").finish(),
             Self::ConnectReady(_) => f.debug_tuple("ConnectReady").finish(),
-            Self::Connected(_) => f.debug_tuple("Connected").finish(),
+            Self::Connected { .. } => f.debug_tuple("Connected").finish(),
             Self::Closed => write!(f, "Closed"),
         }
     }
@@ -145,7 +152,9 @@ impl TcpSocket {
             TcpState::Default(socket) | TcpState::Bound(socket) => {
                 Ok(socket.as_socketlike_view::<std::net::TcpStream>())
             }
-            TcpState::Connected(stream) => Ok(stream.as_socketlike_view::<std::net::TcpStream>()),
+            TcpState::Connected { stream, .. } => {
+                Ok(stream.as_socketlike_view::<std::net::TcpStream>())
+            }
             TcpState::Listening { listener, .. } => {
                 Ok(listener.as_socketlike_view::<std::net::TcpStream>())
             }
@@ -277,9 +286,15 @@ impl TcpSocket {
         match result {
             Ok(stream) => {
                 let stream = Arc::new(stream);
-                self.tcp_state = TcpState::Connected(stream.clone());
-                let input: InputStream = Box::new(TcpReadStream::new(stream.clone()));
-                let output: OutputStream = Box::new(TcpWriteStream::new(stream));
+                let reader = Arc::new(Mutex::new(TcpReader::new(stream.clone())));
+                let writer = Arc::new(Mutex::new(TcpWriter::new(stream.clone())));
+                self.tcp_state = TcpState::Connected {
+                    stream,
+                    reader: reader.clone(),
+                    writer: writer.clone(),
+                };
+                let input: InputStream = Box::new(TcpReadStream(reader));
+                let output: OutputStream = Box::new(TcpWriteStream(writer));
                 Ok((input, output))
             }
             Err(err) => {
@@ -427,9 +442,19 @@ impl TcpSocket {
 
         let client = Arc::new(client);
 
-        let input: InputStream = Box::new(TcpReadStream::new(client.clone()));
-        let output: OutputStream = Box::new(TcpWriteStream::new(client.clone()));
-        let tcp_socket = TcpSocket::from_state(TcpState::Connected(client), self.family)?;
+        let reader = Arc::new(Mutex::new(TcpReader::new(client.clone())));
+        let writer = Arc::new(Mutex::new(TcpWriter::new(client.clone())));
+
+        let input: InputStream = Box::new(TcpReadStream(reader.clone()));
+        let output: OutputStream = Box::new(TcpWriteStream(writer.clone()));
+        let tcp_socket = TcpSocket::from_state(
+            TcpState::Connected {
+                stream: client,
+                reader,
+                writer,
+            },
+            self.family,
+        )?;
 
         Ok((tcp_socket, input, output))
     }
@@ -446,7 +471,7 @@ impl TcpSocket {
 
     pub fn remote_address(&self) -> SocketResult<SocketAddr> {
         let view = match self.tcp_state {
-            TcpState::Connected(..) => self.as_std_view()?,
+            TcpState::Connected { .. } => self.as_std_view()?,
             TcpState::Connecting(..) | TcpState::ConnectReady(..) => {
                 return Err(ErrorCode::ConcurrencyConflict.into())
             }
@@ -613,20 +638,19 @@ impl TcpSocket {
         Ok(())
     }
 
-    pub fn shutdown(&self, how: std::net::Shutdown) -> io::Result<()> {
-        let stream = match &self.tcp_state {
-            TcpState::Connected(stream) => stream,
-            _ => {
-                return Err(io::Error::new(
-                    io::ErrorKind::NotConnected,
-                    "socket not connected",
-                ))
-            }
+    pub fn shutdown(&self, how: Shutdown) -> SocketResult<()> {
+        let TcpState::Connected { reader, writer, .. } = &self.tcp_state else {
+            return Err(ErrorCode::InvalidState.into());
         };
 
-        stream
-            .as_socketlike_view::<std::net::TcpStream>()
-            .shutdown(how)?;
+        if let Shutdown::Both | Shutdown::Read = how {
+            try_lock_for_socket(reader)?.shutdown();
+        }
+
+        if let Shutdown::Both | Shutdown::Write = how {
+            try_lock_for_socket(writer)?.shutdown();
+        }
+
         Ok(())
     }
 }
@@ -641,7 +665,7 @@ impl Subscribe for TcpSocket {
             | TcpState::ListenStarted(..)
             | TcpState::ConnectReady(..)
             | TcpState::Closed
-            | TcpState::Connected(..) => {
+            | TcpState::Connected { .. } => {
                 // No async operation in progress.
             }
             TcpState::Connecting(future) => {
@@ -664,22 +688,18 @@ impl Subscribe for TcpSocket {
     }
 }
 
-struct TcpReadStream {
+struct TcpReader {
     stream: Arc<tokio::net::TcpStream>,
     closed: bool,
 }
 
-impl TcpReadStream {
+impl TcpReader {
     fn new(stream: Arc<tokio::net::TcpStream>) -> Self {
         Self {
             stream,
             closed: false,
         }
     }
-}
-
-#[async_trait::async_trait]
-impl HostInputStream for TcpReadStream {
     fn read(&mut self, size: usize) -> Result<bytes::Bytes, StreamError> {
         if self.closed {
             return Err(StreamError::Closed);
@@ -710,37 +730,57 @@ impl HostInputStream for TcpReadStream {
         buf.truncate(n);
         Ok(buf.freeze())
     }
-}
 
-#[async_trait::async_trait]
-impl Subscribe for TcpReadStream {
+    fn shutdown(&mut self) {
+        native_shutdown(&self.stream, Shutdown::Read);
+        self.closed = true;
+    }
+
     async fn ready(&mut self) {
         if self.closed {
             return;
         }
+
         self.stream.readable().await.unwrap();
     }
 }
 
+struct TcpReadStream(Arc<Mutex<TcpReader>>);
+
+#[async_trait::async_trait]
+impl HostInputStream for TcpReadStream {
+    fn read(&mut self, size: usize) -> Result<bytes::Bytes, StreamError> {
+        try_lock_for_stream(&self.0)?.read(size)
+    }
+}
+
+#[async_trait::async_trait]
+impl Subscribe for TcpReadStream {
+    async fn ready(&mut self) {
+        self.0.lock().await.ready().await
+    }
+}
+
 const SOCKET_READY_SIZE: usize = 1024 * 1024 * 1024;
 
-struct TcpWriteStream {
+struct TcpWriter {
     stream: Arc<tokio::net::TcpStream>,
-    last_write: LastWrite,
+    state: WriteState,
 }
 
-enum LastWrite {
-    Waiting(AbortOnDropJoinHandle<Result<()>>),
-    Error(Error),
-    Done,
+enum WriteState {
+    Ready,
+    Writing(AbortOnDropJoinHandle<io::Result<()>>),
+    Closing(AbortOnDropJoinHandle<io::Result<()>>),
     Closed,
+    Error(io::Error),
 }
 
-impl TcpWriteStream {
+impl TcpWriter {
     fn new(stream: Arc<tokio::net::TcpStream>) -> Self {
         Self {
             stream,
-            last_write: LastWrite::Done,
+            state: WriteState::Ready,
         }
     }
 
@@ -761,10 +801,10 @@ impl TcpWriteStream {
     /// Write `bytes` in a background task, remembering the task handle for use in a future call to
     /// `write_ready`
     fn background_write(&mut self, mut bytes: bytes::Bytes) {
-        assert!(matches!(self.last_write, LastWrite::Done));
+        assert!(matches!(self.state, WriteState::Ready));
 
         let stream = self.stream.clone();
-        self.last_write = LastWrite::Waiting(crate::runtime::spawn(async move {
+        self.state = WriteState::Writing(crate::runtime::spawn(async move {
             // Note: we are not using the AsyncWrite impl here, and instead using the TcpStream
             // primitive try_write, which goes directly to attempt a write with mio. This has
             // two advantages: 1. this operation takes a &TcpStream instead of a &mut TcpStream
@@ -784,14 +824,12 @@ impl TcpWriteStream {
             Ok(())
         }));
     }
-}
 
-#[async_trait::async_trait]
-impl HostOutputStream for TcpWriteStream {
     fn write(&mut self, mut bytes: bytes::Bytes) -> Result<(), StreamError> {
-        match self.last_write {
-            LastWrite::Done => {}
-            LastWrite::Waiting(_) | LastWrite::Error(_) | LastWrite::Closed => {
+        match self.state {
+            WriteState::Ready => {}
+            WriteState::Closed => return Err(StreamError::Closed),
+            WriteState::Writing(_) | WriteState::Closing(_) | WriteState::Error(_) => {
                 return Err(StreamError::Trap(anyhow::anyhow!(
                     "unpermitted: must call check_write first"
                 )));
@@ -812,7 +850,7 @@ impl HostOutputStream for TcpWriteStream {
                 }
 
                 Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
-                    self.last_write = LastWrite::Closed;
+                    self.state = WriteState::Closed;
                     return Err(StreamError::Closed);
                 }
 
@@ -827,23 +865,30 @@ impl HostOutputStream for TcpWriteStream {
         // `flush` is a no-op here, as we're not managing any internal buffer. Additionally,
         // `write_ready` will join the background write task if it's active, so following `flush`
         // with `write_ready` will have the desired effect.
-        match self.last_write {
-            LastWrite::Done | LastWrite::Waiting(_) | LastWrite::Error(_) => Ok(()),
-            LastWrite::Closed => Err(StreamError::Closed),
+        match self.state {
+            WriteState::Ready
+            | WriteState::Writing(_)
+            | WriteState::Closing(_)
+            | WriteState::Error(_) => Ok(()),
+            WriteState::Closed => Err(StreamError::Closed),
         }
     }
 
     fn check_write(&mut self) -> Result<usize, StreamError> {
-        match mem::replace(&mut self.last_write, LastWrite::Closed) {
-            LastWrite::Waiting(task) => {
-                self.last_write = LastWrite::Waiting(task);
+        match mem::replace(&mut self.state, WriteState::Closed) {
+            WriteState::Writing(task) => {
+                self.state = WriteState::Writing(task);
                 return Ok(0);
             }
-            LastWrite::Done => {
-                self.last_write = LastWrite::Done;
+            WriteState::Closing(task) => {
+                self.state = WriteState::Closing(task);
+                return Ok(0);
+            }
+            WriteState::Ready => {
+                self.state = WriteState::Ready;
             }
-            LastWrite::Closed => return Err(StreamError::Closed),
-            LastWrite::Error(e) => return Err(StreamError::LastOperationFailed(e.into())),
+            WriteState::Closed => return Err(StreamError::Closed),
+            WriteState::Error(e) => return Err(StreamError::LastOperationFailed(e.into())),
         }
 
         let writable = self.stream.writable();
@@ -853,25 +898,103 @@ impl HostOutputStream for TcpWriteStream {
         }
         Ok(SOCKET_READY_SIZE)
     }
+
+    fn shutdown(&mut self) {
+        self.state = match mem::replace(&mut self.state, WriteState::Closed) {
+            // No write in progress, immediately shut down:
+            WriteState::Ready => {
+                native_shutdown(&self.stream, Shutdown::Write);
+                WriteState::Closed
+            }
+
+            // Schedule the shutdown after the current write has finished:
+            WriteState::Writing(write) => {
+                let stream = self.stream.clone();
+                WriteState::Closing(crate::runtime::spawn(async move {
+                    let result = write.await;
+                    native_shutdown(&stream, Shutdown::Write);
+                    result
+                }))
+            }
+
+            s => s,
+        };
+    }
+
     async fn cancel(&mut self) {
-        match mem::replace(&mut self.last_write, LastWrite::Closed) {
-            LastWrite::Waiting(task) => _ = task.abort_wait().await,
+        match mem::replace(&mut self.state, WriteState::Closed) {
+            WriteState::Writing(task) | WriteState::Closing(task) => _ = task.cancel().await,
             _ => {}
         }
     }
-}
 
-#[async_trait::async_trait]
-impl Subscribe for TcpWriteStream {
     async fn ready(&mut self) {
-        if let LastWrite::Waiting(task) = &mut self.last_write {
-            self.last_write = match task.await {
-                Ok(()) => LastWrite::Done,
-                Err(e) => LastWrite::Error(e),
-            };
+        match &mut self.state {
+            WriteState::Writing(task) => {
+                self.state = match task.await {
+                    Ok(()) => WriteState::Ready,
+                    Err(e) => WriteState::Error(e),
+                }
+            }
+            WriteState::Closing(task) => {
+                self.state = match task.await {
+                    Ok(()) => WriteState::Closed,
+                    Err(e) => WriteState::Error(e),
+                }
+            }
+            _ => {}
         }
-        if let LastWrite::Done = self.last_write {
+
+        if let WriteState::Ready = self.state {
             self.stream.writable().await.unwrap();
         }
     }
 }
+
+struct TcpWriteStream(Arc<Mutex<TcpWriter>>);
+
+#[async_trait::async_trait]
+impl HostOutputStream for TcpWriteStream {
+    fn write(&mut self, bytes: bytes::Bytes) -> Result<(), StreamError> {
+        try_lock_for_stream(&self.0)?.write(bytes)
+    }
+
+    fn flush(&mut self) -> Result<(), StreamError> {
+        try_lock_for_stream(&self.0)?.flush()
+    }
+
+    fn check_write(&mut self) -> Result<usize, StreamError> {
+        try_lock_for_stream(&self.0)?.check_write()
+    }
+
+    async fn cancel(&mut self) {
+        self.0.lock().await.cancel().await
+    }
+}
+
+#[async_trait::async_trait]
+impl Subscribe for TcpWriteStream {
+    async fn ready(&mut self) {
+        self.0.lock().await.ready().await
+    }
+}
+
+fn native_shutdown(stream: &tokio::net::TcpStream, how: Shutdown) {
+    _ = stream
+        .as_socketlike_view::<std::net::TcpStream>()
+        .shutdown(how);
+}
+
+fn try_lock_for_stream<T>(mutex: &Mutex<T>) -> Result<tokio::sync::MutexGuard<'_, T>, StreamError> {
+    mutex
+        .try_lock()
+        .map_err(|_| StreamError::trap("concurrent access to resource not supported"))
+}
+
+fn try_lock_for_socket<T>(mutex: &Mutex<T>) -> Result<tokio::sync::MutexGuard<'_, T>, SocketError> {
+    mutex.try_lock().map_err(|_| {
+        SocketError::trap(anyhow::anyhow!(
+            "concurrent access to resource not supported"
+        ))
+    })
+}
diff --git a/crates/wasi/src/write_stream.rs b/crates/wasi/src/write_stream.rs
index 33ae91048..fe3658662 100644
--- a/crates/wasi/src/write_stream.rs
+++ b/crates/wasi/src/write_stream.rs
@@ -198,7 +198,7 @@ impl HostOutputStream for AsyncWriteStream {
 
     async fn cancel(&mut self) {
         match self.join_handle.take() {
-            Some(task) => _ = task.abort_wait().await,
+            Some(task) => _ = task.cancel().await,
             None => {}
         }
     }
diff --git a/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs
index a9f462627..171cf010a 100644
--- a/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs
@@ -67,7 +67,7 @@ impl Mmap {
 
     #[inline]
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(
@@ -102,7 +102,7 @@ impl Drop for Mmap {
     fn drop(&mut self) {
         unsafe {
             let ptr = self.memory.as_ptr().cast();
-            let len = self.memory.as_ptr().len();
+            let len = (*self.memory.as_ptr()).len();
             if len == 0 {
                 return;
             }
diff --git a/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs
index b4f834f3a..de2a6aa3f 100644
--- a/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs
@@ -68,7 +68,7 @@ impl Mmap {
     }
 
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(
diff --git a/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs
index ad6f230e1..4c29a9ae1 100644
--- a/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs
@@ -104,7 +104,7 @@ impl Mmap {
 
     #[inline]
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(
@@ -149,7 +149,7 @@ impl Drop for Mmap {
     fn drop(&mut self) {
         unsafe {
             let ptr = self.memory.as_ptr().cast();
-            let len = self.memory.as_ptr().len();
+            let len = (*self.memory.as_ptr()).len();
             if len == 0 {
                 return;
             }
diff --git a/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs
index c0e6bc0ef..d4d62061e 100644
--- a/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs
@@ -164,7 +164,7 @@ impl Mmap {
 
     #[inline]
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(
diff --git a/pulley/src/disas.rs b/pulley/src/disas.rs
index 912b1756d..fc2ab7dbb 100644
--- a/pulley/src/disas.rs
+++ b/pulley/src/disas.rs
@@ -19,6 +19,8 @@ pub struct Disassembler<'a> {
     disas: String,
     start: usize,
     temp: String,
+    offsets: bool,
+    hexdump: bool,
 }
 
 impl<'a> Disassembler<'a> {
@@ -38,9 +40,27 @@ impl<'a> Disassembler<'a> {
             disas: String::new(),
             start: 0,
             temp: String::new(),
+            offsets: true,
+            hexdump: true,
         }
     }
 
+    /// Whether to prefix each instruction's disassembly with its offset.
+    ///
+    /// True by default.
+    pub fn offsets(&mut self, offsets: bool) -> &mut Self {
+        self.offsets = offsets;
+        self
+    }
+
+    /// Whether to include a hexdump of the bytecode in the disassembly.
+    ///
+    /// True by default.
+    pub fn hexdump(&mut self, hexdump: bool) -> &mut Self {
+        self.hexdump = hexdump;
+        self
+    }
+
     /// Get the disassembly thus far.
     pub fn disas(&self) -> &str {
         &self.disas
@@ -176,18 +196,21 @@ macro_rules! impl_disas {
             }
 
             fn after_visit(&mut self) {
-                let size = self.bytecode.position() - self.start;
-
-                write!(&mut self.disas, "{:8x}: ", self.start).unwrap();
-                let mut need_space = false;
-                for byte in &self.raw_bytecode[self.start..][..size] {
-                    write!(&mut self.disas, "{}{byte:02x}", if need_space { " " } else { "" }).unwrap();
-                    need_space = true;
+                if self.offsets {
+                    write!(&mut self.disas, "{:8x}: ", self.start).unwrap();
                 }
-                for _ in 0..11_usize.saturating_sub(size) {
-                    write!(&mut self.disas, "   ").unwrap();
+                if self.hexdump {
+                    let size = self.bytecode.position() - self.start;
+                    let mut need_space = false;
+                    for byte in &self.raw_bytecode[self.start..][..size] {
+                        let space = if need_space { " " } else { "" };
+                        write!(&mut self.disas, "{}{byte:02x}", space).unwrap();
+                        need_space = true;
+                    }
+                    for _ in 0..11_usize.saturating_sub(size) {
+                        write!(&mut self.disas, "   ").unwrap();
+                    }
                 }
-
                 self.disas.push_str(&self.temp);
                 self.temp.clear();
 


Test patch:
diff --git a/crates/test-programs/src/bin/preview2_tcp_streams.rs b/crates/test-programs/src/bin/preview2_tcp_streams.rs
index 688827be6..c75695896 100644
--- a/crates/test-programs/src/bin/preview2_tcp_streams.rs
+++ b/crates/test-programs/src/bin/preview2_tcp_streams.rs
@@ -25,6 +25,31 @@ fn test_tcp_input_stream_should_be_closed_by_remote_shutdown(
     });
 }
 
+/// InputStream::read should return `StreamError::Closed` after the connection has been shut down locally.
+fn test_tcp_input_stream_should_be_closed_by_local_shutdown(
+    net: &Network,
+    family: IpAddressFamily,
+) {
+    setup(net, family, |server, client| {
+        // On Linux, `recv` continues to work even after `shutdown(sock, SHUT_RD)`
+        // has been called. To properly test that this behavior doesn't happen in
+        // WASI, we make sure there's some data to read by the client:
+        server.output.blocking_write_util(b"Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.").unwrap();
+
+        // Wait for the data to reach the client:
+        client.input.subscribe().block();
+
+        // Shut down socket locally:
+        client.socket.shutdown(ShutdownType::Receive).unwrap();
+
+        // The input stream should immediately signal StreamError::Closed.
+        assert!(matches!(client.input.read(10), Err(StreamError::Closed)));
+
+        // Stream should still be closed, even when requesting 0 bytes:
+        assert!(matches!(client.input.read(0), Err(StreamError::Closed)));
+    });
+}
+
 /// OutputStream should return `StreamError::Closed` after the connection has been locally shut down for sending.
 fn test_tcp_output_stream_should_be_closed_by_local_shutdown(
     net: &Network,
@@ -54,14 +79,50 @@ fn test_tcp_output_stream_should_be_closed_by_local_shutdown(
     });
 }
 
+/// Calling `shutdown` while the OutputStream is in the middle of a background write should not cause that write to be lost.
+fn test_tcp_shutdown_should_not_lose_data(net: &Network, family: IpAddressFamily) {
+    setup(net, family, |server, client| {
+        // Minimize the local send buffer:
+        client.socket.set_send_buffer_size(1024).unwrap();
+        let small_buffer_size = client.socket.send_buffer_size().unwrap();
+
+        // Create a significantly bigger buffer, so that we can be pretty sure the `write` won't finish immediately:
+        let big_buffer_size = client
+            .output
+            .check_write()
+            .unwrap()
+            .min(100 * small_buffer_size);
+        assert!(big_buffer_size > small_buffer_size);
+        let outgoing_data = vec![0; big_buffer_size as usize];
+
+        // Submit the oversized buffer and immediately initiate the shutdown:
+        client.output.write(&outgoing_data).unwrap();
+        client.socket.shutdown(ShutdownType::Send).unwrap();
+
+        // The peer should receive _all_ data:
+        let incoming_data = server.input.blocking_read_to_end().unwrap();
+        assert_eq!(
+            outgoing_data.len(),
+            incoming_data.len(),
+            "Received data should match the sent data"
+        );
+    });
+}
+
 fn main() {
     let net = Network::default();
 
     test_tcp_input_stream_should_be_closed_by_remote_shutdown(&net, IpAddressFamily::Ipv4);
     test_tcp_input_stream_should_be_closed_by_remote_shutdown(&net, IpAddressFamily::Ipv6);
 
+    test_tcp_input_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv4);
+    test_tcp_input_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv6);
+
     test_tcp_output_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv4);
     test_tcp_output_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv6);
+
+    test_tcp_shutdown_should_not_lose_data(&net, IpAddressFamily::Ipv4);
+    test_tcp_shutdown_should_not_lose_data(&net, IpAddressFamily::Ipv6);
 }
 
 struct Connection {
diff --git a/crates/test-programs/src/sockets.rs b/crates/test-programs/src/sockets.rs
index c77ad8d7f..0fcccaaab 100644
--- a/crates/test-programs/src/sockets.rs
+++ b/crates/test-programs/src/sockets.rs
@@ -29,6 +29,19 @@ impl Pollable {
     }
 }
 
+impl InputStream {
+    pub fn blocking_read_to_end(&self) -> Result<Vec<u8>, crate::wasi::io::error::Error> {
+        let mut data = vec![];
+        loop {
+            match self.blocking_read(1024 * 1024) {
+                Ok(chunk) => data.extend(chunk),
+                Err(StreamError::Closed) => return Ok(data),
+                Err(StreamError::LastOperationFailed(e)) => return Err(e),
+            }
+        }
+    }
+}
+
 impl OutputStream {
     pub fn blocking_write_util(&self, mut bytes: &[u8]) -> Result<(), StreamError> {
         let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS);
diff --git a/pulley/tests/all/disas.rs b/pulley/tests/all/disas.rs
index eb2b49a56..5b9c1956c 100644
--- a/pulley/tests/all/disas.rs
+++ b/pulley/tests/all/disas.rs
@@ -11,19 +11,25 @@ fn encoded(ops: &[Op]) -> Vec<u8> {
     encoded
 }
 
-fn assert_disas(ops: &[Op], expected: &str) {
+#[track_caller]
+fn assert_disas_with_disassembler(dis: &mut disas::Disassembler<'_>, expected: &str) {
     let expected = expected.trim();
     eprintln!("=== expected ===\n{expected}");
 
-    let bytecode = encoded(ops);
+    decode::Decoder::decode_all(dis).expect("decoding should succeed");
 
-    let actual = disas::Disassembler::disassemble_all(&bytecode).expect("decoding failed");
-    let actual = actual.trim();
+    let actual = dis.disas().trim();
     eprintln!("=== actual ===\n{actual}");
 
     assert_eq!(expected, actual);
 }
 
+#[track_caller]
+fn assert_disas(ops: &[Op], expected: &str) {
+    let bytecode = encoded(ops);
+    assert_disas_with_disassembler(&mut disas::Disassembler::new(&bytecode), expected);
+}
+
 #[test]
 fn simple() {
     assert_disas(
@@ -85,3 +91,92 @@ fn push_pop_many() {
         "#,
     );
 }
+
+#[test]
+fn no_offsets() {
+    let bytecode = encoded(&[
+        // Prologue.
+        Op::PushFrame(PushFrame {}),
+        // Function body.
+        Op::Xadd32(Xadd32 {
+            operands: BinaryOperands {
+                dst: XReg::x0,
+                src1: XReg::x0,
+                src2: XReg::x1,
+            },
+        }),
+        // Epilogue.
+        Op::PopFrame(PopFrame {}),
+        Op::Ret(Ret {}),
+    ]);
+
+    assert_disas_with_disassembler(
+        disas::Disassembler::new(&bytecode).offsets(false),
+        r#"
+2f                              push_frame
+12 00 04                        xadd32 x0, x0, x1
+30                              pop_frame
+00                              ret
+        "#,
+    );
+}
+
+#[test]
+fn no_hexdump() {
+    let bytecode = encoded(&[
+        // Prologue.
+        Op::PushFrame(PushFrame {}),
+        // Function body.
+        Op::Xadd32(Xadd32 {
+            operands: BinaryOperands {
+                dst: XReg::x0,
+                src1: XReg::x0,
+                src2: XReg::x1,
+            },
+        }),
+        // Epilogue.
+        Op::PopFrame(PopFrame {}),
+        Op::Ret(Ret {}),
+    ]);
+
+    assert_disas_with_disassembler(
+        disas::Disassembler::new(&bytecode).hexdump(false),
+        r#"
+       0: push_frame
+       1: xadd32 x0, x0, x1
+       4: pop_frame
+       5: ret
+        "#,
+    );
+}
+
+#[test]
+fn no_offsets_or_hexdump() {
+    let bytecode = encoded(&[
+        // Prologue.
+        Op::PushFrame(PushFrame {}),
+        // Function body.
+        Op::Xadd32(Xadd32 {
+            operands: BinaryOperands {
+                dst: XReg::x0,
+                src1: XReg::x0,
+                src2: XReg::x1,
+            },
+        }),
+        // Epilogue.
+        Op::PopFrame(PopFrame {}),
+        Op::Ret(Ret {}),
+    ]);
+
+    assert_disas_with_disassembler(
+        disas::Disassembler::new(&bytecode)
+            .offsets(false)
+            .hexdump(false),
+        r#"
+push_frame
+xadd32 x0, x0, x1
+pop_frame
+ret
+        "#,
+    );
+}