Switch Subtask trait to taking &self (#1390)

This enables having a subtask dispatched based on in-memory variable
information as opposed to requiring static information up-front. I'm
intending to use this for `wit-dylib` over in wasm-tools to prove out
some async bits. This'll eventually happen for futures/streams most
likely as well, but for now it's just subtasks.

This also shouldn't affect runtime code generated for this crate from
before since the types used are all zero-sized and will get elided at
runtime. Thus this is intended to enable more use cases without altering
existing ones.

Author: alexcrichton

crates/guest-rust/src/lib.rs

@@ -872,10 +872,12 @@ pub mod examples;
 #[doc(hidden)]
 pub mod rt;
 
+#[cfg(feature = "async")]
+#[allow(deprecated)]
+pub use rt::async_support::backpressure_set;
 #[cfg(feature = "async")]
 pub use rt::async_support::{
-    backpressure_dec, backpressure_inc, backpressure_set, block_on, spawn, yield_async,
-    yield_blocking, AbiBuffer, FutureRead, FutureReader, FutureWrite, FutureWriteCancel,
-    FutureWriteError, FutureWriter, StreamRead, StreamReader, StreamResult, StreamWrite,
-    StreamWriter,
+    backpressure_dec, backpressure_inc, block_on, spawn, yield_async, yield_blocking, AbiBuffer,
+    FutureRead, FutureReader, FutureWrite, FutureWriteCancel, FutureWriteError, FutureWriter,
+    StreamRead, StreamReader, StreamResult, StreamWrite, StreamWriter,
 };

crates/guest-rust/src/rt/async_support/future_support.rs

@@ -269,7 +269,7 @@ impl<T> FutureWriter<T> {
     /// In such a situation the operation can be retried at a future date.
     pub fn write(self, value: T) -> FutureWrite<T> {
         FutureWrite {
-            op: WaitableOperation::new((self, value)),
+            op: WaitableOperation::new(FutureWriteOp(marker::PhantomData), (self, value)),
         }
     }
 }
@@ -339,7 +339,7 @@ where
     type Result = (WriteComplete<T>, FutureWriter<T>);
     type Cancel = FutureWriteCancel<T>;
 
-    fn start((writer, value): Self::Start) -> (u32, Self::InProgress) {
+    fn start(&self, (writer, value): Self::Start) -> (u32, Self::InProgress) {
         // TODO: it should be safe to store the lower-destination in
         // `WaitableOperation` using `Pin` memory and such, but that would
         // require some type-level trickery to get a correctly-sized value
@@ -361,11 +361,12 @@ where
         (code, (writer, cleanup))
     }
 
-    fn start_cancelled((writer, value): Self::Start) -> Self::Cancel {
+    fn start_cancelled(&self, (writer, value): Self::Start) -> Self::Cancel {
         FutureWriteCancel::Cancelled(value, writer)
     }
 
     fn in_progress_update(
+        &self,
         (mut writer, cleanup): Self::InProgress,
         code: u32,
     ) -> Result<Self::Result, Self::InProgress> {
@@ -425,19 +426,19 @@ where
         }
     }
 
-    fn in_progress_waitable((writer, _): &Self::InProgress) -> u32 {
+    fn in_progress_waitable(&self, (writer, _): &Self::InProgress) -> u32 {
         writer.handle
     }
 
-    fn in_progress_cancel((writer, _): &Self::InProgress) -> u32 {
+    fn in_progress_cancel(&self, (writer, _): &Self::InProgress) -> u32 {
         // SAFETY: we're managing `writer` and all the various operational bits,
         // so this relies on `WaitableOperation` being safe.
         let code = unsafe { (writer.vtable.cancel_write)(writer.handle) };
         rtdebug!("future.cancel-write({}) = {code:#x}", writer.handle);
         code
     }
 
-    fn result_into_cancel((result, writer): Self::Result) -> Self::Cancel {
+    fn result_into_cancel(&self, (result, writer): Self::Result) -> Self::Cancel {
         match result {
             // The value was actually sent, meaning we can't yield back the
             // future nor the value.
@@ -581,7 +582,7 @@ impl<T> IntoFuture for FutureReader<T> {
     /// written to the writable end of this `future`.
     fn into_future(self) -> Self::IntoFuture {
         FutureRead {
-            op: WaitableOperation::new(self),
+            op: WaitableOperation::new(FutureReadOp(marker::PhantomData), self),
         }
     }
 }
@@ -622,7 +623,7 @@ where
     type Result = (ReadComplete<T>, FutureReader<T>);
     type Cancel = Result<T, FutureReader<T>>;
 
-    fn start(reader: Self::Start) -> (u32, Self::InProgress) {
+    fn start(&self, reader: Self::Start) -> (u32, Self::InProgress) {
         let (ptr, cleanup) = Cleanup::new(reader.vtable.layout);
         // SAFETY: `ptr` is allocated with `vtable.layout` and should be
         // safe to use here. Its lifetime for the async operation is hinged on
@@ -632,11 +633,12 @@ where
         (code, (reader, cleanup))
     }
 
-    fn start_cancelled(state: Self::Start) -> Self::Cancel {
+    fn start_cancelled(&self, state: Self::Start) -> Self::Cancel {
         Err(state)
     }
 
     fn in_progress_update(
+        &self,
         (reader, cleanup): Self::InProgress,
         code: u32,
     ) -> Result<Self::Result, Self::InProgress> {
@@ -667,19 +669,19 @@ where
         }
     }
 
-    fn in_progress_waitable((reader, _): &Self::InProgress) -> u32 {
+    fn in_progress_waitable(&self, (reader, _): &Self::InProgress) -> u32 {
         reader.handle()
     }
 
-    fn in_progress_cancel((reader, _): &Self::InProgress) -> u32 {
+    fn in_progress_cancel(&self, (reader, _): &Self::InProgress) -> u32 {
         // SAFETY: we're managing `reader` and all the various operational bits,
         // so this relies on `WaitableOperation` being safe.
         let code = unsafe { (reader.vtable.cancel_read)(reader.handle()) };
         rtdebug!("future.cancel-read({}) = {code:#x}", reader.handle());
         code
     }
 
-    fn result_into_cancel((value, reader): Self::Result) -> Self::Cancel {
+    fn result_into_cancel(&self, (value, reader): Self::Result) -> Self::Cancel {
         match value {
             // The value was actually read, so thread that through here.
             ReadComplete::Value(value) => Ok(value),

crates/guest-rust/src/rt/async_support/stream_support.rs

@@ -143,7 +143,7 @@ impl<T> StreamWriter<T> {
     /// instead of `Vec<T>`.
     pub fn write_buf(&mut self, values: AbiBuffer<T>) -> StreamWrite<'_, T> {
         StreamWrite {
-            op: WaitableOperation::new((self, values)),
+            op: WaitableOperation::new(StreamWriteOp(marker::PhantomData), (self, values)),
         }
     }
 
@@ -243,7 +243,7 @@ where
     type Result = (StreamResult, AbiBuffer<T>);
     type Cancel = (StreamResult, AbiBuffer<T>);
 
-    fn start((writer, buf): Self::Start) -> (u32, Self::InProgress) {
+    fn start(&self, (writer, buf): Self::Start) -> (u32, Self::InProgress) {
         if writer.done {
             return (DROPPED, (writer, buf));
         }
@@ -259,11 +259,12 @@ where
         (code, (writer, buf))
     }
 
-    fn start_cancelled((_writer, buf): Self::Start) -> Self::Cancel {
+    fn start_cancelled(&self, (_writer, buf): Self::Start) -> Self::Cancel {
         (StreamResult::Cancelled, buf)
     }
 
     fn in_progress_update(
+        &self,
         (writer, mut buf): Self::InProgress,
         code: u32,
     ) -> Result<Self::Result, Self::InProgress> {
@@ -284,19 +285,19 @@ where
         }
     }
 
-    fn in_progress_waitable((writer, _): &Self::InProgress) -> u32 {
+    fn in_progress_waitable(&self, (writer, _): &Self::InProgress) -> u32 {
         writer.handle
     }
 
-    fn in_progress_cancel((writer, _): &Self::InProgress) -> u32 {
+    fn in_progress_cancel(&self, (writer, _): &Self::InProgress) -> u32 {
         // SAFETY: we're managing `writer` and all the various operational bits,
         // so this relies on `WaitableOperation` being safe.
         let code = unsafe { (writer.vtable.cancel_write)(writer.handle) };
         rtdebug!("stream.cancel-write({}) = {code:#x}", writer.handle);
         code
     }
 
-    fn result_into_cancel(result: Self::Result) -> Self::Cancel {
+    fn result_into_cancel(&self, result: Self::Result) -> Self::Cancel {
         result
     }
 }
@@ -391,7 +392,7 @@ impl<T> StreamReader<T> {
     /// used.
     pub fn read(&mut self, buf: Vec<T>) -> StreamRead<'_, T> {
         StreamRead {
-            op: WaitableOperation::new((self, buf)),
+            op: WaitableOperation::new(StreamReadOp(marker::PhantomData), (self, buf)),
         }
     }
 
@@ -459,7 +460,7 @@ where
     type Result = (StreamResult, Vec<T>);
     type Cancel = (StreamResult, Vec<T>);
 
-    fn start((reader, mut buf): Self::Start) -> (u32, Self::InProgress) {
+    fn start(&self, (reader, mut buf): Self::Start) -> (u32, Self::InProgress) {
         if reader.done {
             return (DROPPED, (reader, buf, None));
         }
@@ -492,11 +493,12 @@ where
         (code, (reader, buf, cleanup))
     }
 
-    fn start_cancelled((_, buf): Self::Start) -> Self::Cancel {
+    fn start_cancelled(&self, (_, buf): Self::Start) -> Self::Cancel {
         (StreamResult::Cancelled, buf)
     }
 
     fn in_progress_update(
+        &self,
         (reader, mut buf, cleanup): Self::InProgress,
         code: u32,
     ) -> Result<Self::Result, Self::InProgress> {
@@ -553,19 +555,19 @@ where
         }
     }
 
-    fn in_progress_waitable((reader, ..): &Self::InProgress) -> u32 {
+    fn in_progress_waitable(&self, (reader, ..): &Self::InProgress) -> u32 {
         reader.handle()
     }
 
-    fn in_progress_cancel((reader, ..): &Self::InProgress) -> u32 {
+    fn in_progress_cancel(&self, (reader, ..): &Self::InProgress) -> u32 {
         // SAFETY: we're managing `reader` and all the various operational bits,
         // so this relies on `WaitableOperation` being safe.
         let code = unsafe { (reader.vtable.cancel_read)(reader.handle()) };
         rtdebug!("stream.cancel-read({}) = {code:#x}", reader.handle());
         code
     }
 
-    fn result_into_cancel(result: Self::Result) -> Self::Cancel {
+    fn result_into_cancel(&self, result: Self::Result) -> Self::Cancel {
         result
     }
 }