Refine Guard* APIs and implementation (#11354)

This is a follow-up to #11325 with a number of cosmetic changes about
the shape of the API and structure of the internals:

* `{Stream,Future}{Reader,Writer}::guard` is now an alternative
  constructor to `Guard*::new` (import fewer types).
* Internally `WithAccessor` and `DropWithStore` are removed in favor of
  direct `Drop for Guard*` impls.
* An `Option` is used to replace `ManuallyDrop` and `unsafe` code.
* `{Stream,Future}{Reader,Writer}::close{,_with}` now take `&mut self`
  instead of `self` to be more composable with `&mut self` arguments
  during `Drop` for other structures (e.g. build-your-own
  drop-with-store).
* The type parameters on `Guard*` are simplified to just `T`, the future
  or stream payload, and `A: AsAccessor`. This helps cut down on the
  complexity of signatures.
* `Guard*` types now have `into_{stream,future}` as an alternative to
  `.into()` which doesn't require type annotations.

Author: alexcrichton

crates/misc/component-async-tests/tests/scenario/streams.rs

@@ -15,8 +15,8 @@ use {
     wasmtime::{
         Engine, Store,
         component::{
-            GuardedFutureReader, GuardedStreamReader, GuardedStreamWriter, Linker, ResourceTable,
-            VecBuffer,
+            Accessor, GuardedFutureReader, GuardedStreamReader, GuardedStreamWriter, Linker,
+            ResourceTable, VecBuffer,
         },
     },
     wasmtime_wasi::p2::WasiCtxBuilder,
@@ -49,15 +49,15 @@ pub async fn async_watch_streams() -> Result<()> {
     let instance = linker.instantiate_async(&mut store, &component).await?;
 
     // Test watching and then dropping the read end of a stream.
-    let (mut tx, rx) = instance.stream::<u8>(&mut store)?;
+    let (mut tx, mut rx) = instance.stream::<u8>(&mut store)?;
     instance
         .run_concurrent(&mut store, async |store| {
             futures::join!(tx.watch_reader(store), async { rx.close_with(store) }).1
         })
         .await?;
 
     // Test dropping and then watching the read end of a stream.
-    let (mut tx, rx) = instance.stream::<u8>(&mut store)?;
+    let (mut tx, mut rx) = instance.stream::<u8>(&mut store)?;
     instance
         .run_concurrent(&mut store, async |store| {
             rx.close_with(store);
@@ -66,15 +66,15 @@ pub async fn async_watch_streams() -> Result<()> {
         .await?;
 
     // Test watching and then dropping the write end of a stream.
-    let (tx, mut rx) = instance.stream::<u8>(&mut store)?;
+    let (mut tx, mut rx) = instance.stream::<u8>(&mut store)?;
     instance
         .run_concurrent(&mut store, async |store| {
             futures::join!(rx.watch_writer(store), async { tx.close_with(store) }).1
         })
         .await?;
 
     // Test dropping and then watching the write end of a stream.
-    let (tx, mut rx) = instance.stream::<u8>(&mut store)?;
+    let (mut tx, mut rx) = instance.stream::<u8>(&mut store)?;
     instance
         .run_concurrent(&mut store, async |store| {
             tx.close_with(store);
@@ -83,15 +83,15 @@ pub async fn async_watch_streams() -> Result<()> {
         .await?;
 
     // Test watching and then dropping the read end of a future.
-    let (mut tx, rx) = instance.future::<u8>(&mut store, || 42)?;
+    let (mut tx, mut rx) = instance.future::<u8>(&mut store, || 42)?;
     instance
         .run_concurrent(&mut store, async |store| {
             futures::join!(tx.watch_reader(store), async { rx.close_with(store) }).1
         })
         .await?;
 
     // Test dropping and then watching the read end of a future.
-    let (mut tx, rx) = instance.future::<u8>(&mut store, || 42)?;
+    let (mut tx, mut rx) = instance.future::<u8>(&mut store, || 42)?;
     instance
         .run_concurrent(&mut store, async |store| {
             rx.close_with(store);
@@ -100,15 +100,15 @@ pub async fn async_watch_streams() -> Result<()> {
         .await?;
 
     // Test watching and then dropping the write end of a future.
-    let (tx, mut rx) = instance.future::<u8>(&mut store, || 42)?;
+    let (mut tx, mut rx) = instance.future::<u8>(&mut store, || 42)?;
     instance
         .run_concurrent(&mut store, async |store| {
             futures::join!(rx.watch_writer(store), async { tx.close_with(store) }).1
         })
         .await?;
 
     // Test dropping and then watching the write end of a future.
-    let (tx, mut rx) = instance.future::<u8>(&mut store, || 42)?;
+    let (mut tx, mut rx) = instance.future::<u8>(&mut store, || 42)?;
     instance
         .run_concurrent(&mut store, async |store| {
             tx.close_with(store);
@@ -117,8 +117,11 @@ pub async fn async_watch_streams() -> Result<()> {
         .await?;
 
     enum Event<'a> {
-        Write(Option<GuardedStreamWriter<'a, u8, Ctx>>),
-        Read(Option<GuardedStreamReader<'a, u8, Ctx>>, Option<u8>),
+        Write(Option<GuardedStreamWriter<u8, &'a Accessor<Ctx>>>),
+        Read(
+            Option<GuardedStreamReader<u8, &'a Accessor<Ctx>>>,
+            Option<u8>,
+        ),
     }
 
     // Test watching, then writing to, then dropping, then writing again to the
@@ -212,9 +215,9 @@ pub async fn test_closed_streams(watch: bool) -> Result<()> {
     let instance = linker.instantiate_async(&mut store, &component).await?;
 
     enum StreamEvent<'a> {
-        FirstWrite(Option<GuardedStreamWriter<'a, u8, Ctx>>),
-        FirstRead(Option<GuardedStreamReader<'a, u8, Ctx>>, Vec<u8>),
-        SecondWrite(Option<GuardedStreamWriter<'a, u8, Ctx>>),
+        FirstWrite(Option<GuardedStreamWriter<u8, &'a Accessor<Ctx>>>),
+        FirstRead(Option<GuardedStreamReader<u8, &'a Accessor<Ctx>>>, Vec<u8>),
+        SecondWrite(Option<GuardedStreamWriter<u8, &'a Accessor<Ctx>>>),
         GuestCompleted,
     }
 

crates/misc/component-async-tests/tests/scenario/transmit.rs

@@ -364,15 +364,18 @@ async fn test_transmit_with<Test: TransmitTest + 'static>(component: &str) -> Re
 
     enum Event<'a, Test: TransmitTest> {
         Result(Test::Result),
-        ControlWriteA(Option<GuardedStreamWriter<'a, Control, Ctx>>),
-        ControlWriteB(Option<GuardedStreamWriter<'a, Control, Ctx>>),
-        ControlWriteC(Option<GuardedStreamWriter<'a, Control, Ctx>>),
+        ControlWriteA(Option<GuardedStreamWriter<Control, &'a Accessor<Ctx>>>),
+        ControlWriteB(Option<GuardedStreamWriter<Control, &'a Accessor<Ctx>>>),
+        ControlWriteC(Option<GuardedStreamWriter<Control, &'a Accessor<Ctx>>>),
         ControlWriteD,
         WriteA,
         WriteB(bool),
-        ReadC(Option<GuardedStreamReader<'a, String, Ctx>>, Option<String>),
+        ReadC(
+            Option<GuardedStreamReader<String, &'a Accessor<Ctx>>>,
+            Option<String>,
+        ),
         ReadD(Option<String>),
-        ReadNone(Option<GuardedStreamReader<'a, String, Ctx>>),
+        ReadNone(Option<GuardedStreamReader<String, &'a Accessor<Ctx>>>),
     }
 
     let (control_tx, control_rx) = instance.stream(&mut store)?;

crates/wasi/src/p3/sockets/host/types/tcp.rs

@@ -392,7 +392,7 @@ impl HostTcpSocketWithStore for WasiSockets {
     ) -> wasmtime::Result<(StreamReader<u8>, FutureReader<Result<(), ErrorCode>>)> {
         store.with(|mut view| {
             let instance = view.instance();
-            let (data_tx, data_rx) = instance
+            let (mut data_tx, data_rx) = instance
                 .stream(&mut view)
                 .context("failed to create stream")?;
             let TcpSocket { tcp_state, .. } = get_socket_mut(view.get().table, &socket)?;
@@ -411,7 +411,7 @@ impl HostTcpSocketWithStore for WasiSockets {
                 }
                 prev => {
                     *tcp_state = prev;
-                    let (result_tx, result_rx) = instance
+                    let (mut result_tx, result_rx) = instance
                         .future(&mut view, || Err(ErrorCode::InvalidState))
                         .context("failed to create future")?;
                     result_tx.close(&mut view);