async: remove the need for {Ready,Empty}Producer

rvolosatovs · rvolosatovs · commit 4da55948f349 · 2025-09-12T11:46:37.000+02:00
Signed-off-by: Roman Volosatovs &lt;rvolosatovs@riseup.net&gt;
diff --git a/crates/wasi-http/src/p3/body.rs b/crates/wasi-http/src/p3/body.rs
@@ -2,6 +2,7 @@ use crate::p3::bindings::http::types::{ErrorCode, Fields, Trailers};
 use crate::p3::{WasiHttp, WasiHttpCtxView};
 use anyhow::Context as _;
 use bytes::Bytes;
+use core::iter;
 use core::num::NonZeroUsize;
 use core::pin::Pin;
 use core::task::{Context, Poll, ready};
@@ -13,8 +14,8 @@ use std::sync::Arc;
 use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::PollSender;
 use wasmtime::component::{
-    Access, Destination, EmptyProducer, FutureConsumer, FutureReader, Resource, Source,
-    StreamConsumer, StreamProducer, StreamReader, StreamResult,
+    Access, Destination, FutureConsumer, FutureReader, Resource, Source, StreamConsumer,
+    StreamProducer, StreamReader, StreamResult,
 };
 use wasmtime::{AsContextMut, StoreContextMut};
 
@@ -74,7 +75,7 @@ impl Body {
                 // https://github.com/WebAssembly/wasi-http/issues/176
                 _ = result_tx.send(Box::new(async { Ok(()) }));
                 Ok((
-                    StreamReader::new(instance, &mut store, EmptyProducer::default()),
+                    StreamReader::new(instance, &mut store, iter::empty()),
                     trailers_rx,
                 ))
             }
diff --git a/crates/wasi/src/p3/filesystem/host.rs b/crates/wasi/src/p3/filesystem/host.rs
@@ -9,18 +9,18 @@ use crate::p3::{DEFAULT_BUFFER_CAPACITY, FallibleIteratorProducer};
 use crate::{DirPerms, FilePerms};
 use anyhow::Context as _;
 use bytes::BytesMut;
-use core::mem;
 use core::pin::Pin;
 use core::task::{Context, Poll, ready};
+use core::{iter, mem};
 use std::io::{self, Cursor};
 use std::sync::Arc;
 use system_interface::fs::FileIoExt as _;
 use tokio::sync::{mpsc, oneshot};
 use tokio::task::{JoinHandle, spawn_blocking};
 use wasmtime::StoreContextMut;
 use wasmtime::component::{
-    Accessor, Destination, EmptyProducer, FutureReader, ReadyProducer, Resource, ResourceTable,
-    Source, StreamConsumer, StreamProducer, StreamReader, StreamResult,
+    Accessor, Destination, FutureReader, Resource, ResourceTable, Source, StreamConsumer,
+    StreamProducer, StreamReader, StreamResult,
 };
 
 fn get_descriptor<'a>(
@@ -495,12 +495,10 @@ impl types::HostDescriptorWithStore for WasiFilesystem {
             let file = get_file(store.get().table, &fd)?;
             if !file.perms.contains(FilePerms::READ) {
                 return Ok((
-                    StreamReader::new(instance, &mut store, EmptyProducer::default()),
-                    FutureReader::new(
-                        instance,
-                        &mut store,
-                        ReadyProducer::from(Err(ErrorCode::NotPermitted)),
-                    ),
+                    StreamReader::new(instance, &mut store, iter::empty()),
+                    FutureReader::new(instance, &mut store, async {
+                        anyhow::Ok(Err(ErrorCode::NotPermitted))
+                    }),
                 ));
             }
 
@@ -641,12 +639,10 @@ impl types::HostDescriptorWithStore for WasiFilesystem {
             let dir = get_dir(store.get().table, &fd)?;
             if !dir.perms.contains(DirPerms::READ) {
                 return Ok((
-                    StreamReader::new(instance, &mut store, EmptyProducer::default()),
-                    FutureReader::new(
-                        instance,
-                        &mut store,
-                        ReadyProducer::from(Err(ErrorCode::NotPermitted)),
-                    ),
+                    StreamReader::new(instance, &mut store, iter::empty()),
+                    FutureReader::new(instance, &mut store, async {
+                        anyhow::Ok(Err(ErrorCode::NotPermitted))
+                    }),
                 ));
             }
             let allow_blocking_current_thread = dir.allow_blocking_current_thread;
@@ -664,7 +660,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem {
                     ),
                     Err(e) => {
                         result_tx.send(Err(e.into())).unwrap();
-                        StreamReader::new(instance, &mut store, EmptyProducer::default())
+                        StreamReader::new(instance, &mut store, iter::empty())
                     }
                 }
             } else {
diff --git a/crates/wasi/src/p3/sockets/host/types/tcp.rs b/crates/wasi/src/p3/sockets/host/types/tcp.rs
@@ -8,6 +8,7 @@ use crate::p3::sockets::{SocketError, SocketResult, WasiSockets};
 use crate::sockets::{NonInheritedOptions, SocketAddrUse, SocketAddressFamily, WasiSocketsCtxView};
 use anyhow::Context as _;
 use bytes::BytesMut;
+use core::iter;
 use core::pin::Pin;
 use core::task::{Context, Poll};
 use io_lifetimes::AsSocketlike as _;
@@ -17,8 +18,8 @@ use std::sync::Arc;
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::oneshot;
 use wasmtime::component::{
-    Accessor, Destination, EmptyProducer, FutureReader, ReadyProducer, Resource, ResourceTable,
-    Source, StreamConsumer, StreamProducer, StreamReader, StreamResult,
+    Accessor, Destination, FutureReader, Resource, ResourceTable, Source, StreamConsumer,
+    StreamProducer, StreamReader, StreamResult,
 };
 use wasmtime::{AsContextMut as _, StoreContextMut};
 
@@ -353,12 +354,10 @@ impl HostTcpSocketWithStore for WasiSockets {
                     ))
                 }
                 None => Ok((
-                    StreamReader::new(instance, &mut store, EmptyProducer::default()),
-                    FutureReader::new(
-                        instance,
-                        &mut store,
-                        ReadyProducer::from(Err(ErrorCode::InvalidState)),
-                    ),
+                    StreamReader::new(instance, &mut store, iter::empty()),
+                    FutureReader::new(instance, &mut store, async {
+                        anyhow::Ok(Err(ErrorCode::InvalidState))
+                    }),
                 )),
             }
         })
diff --git a/crates/wasmtime/src/runtime/component/concurrent.rs b/crates/wasmtime/src/runtime/component/concurrent.rs
@@ -92,10 +92,9 @@ use wasmtime_environ::component::{
 
 pub use abort::JoinHandle;
 pub use futures_and_streams::{
-    Destination, DirectDestination, DirectSource, EmptyProducer, ErrorContext, FutureConsumer,
-    FutureProducer, FutureReader, GuardedFutureReader, GuardedStreamReader, ReadBuffer,
-    ReadyProducer, Source, StreamConsumer, StreamProducer, StreamReader, StreamResult, VecBuffer,
-    WriteBuffer,
+    Destination, DirectDestination, DirectSource, ErrorContext, FutureConsumer, FutureProducer,
+    FutureReader, GuardedFutureReader, GuardedStreamReader, ReadBuffer, Source, StreamConsumer,
+    StreamProducer, StreamReader, StreamResult, VecBuffer, WriteBuffer,
 };
 pub(crate) use futures_and_streams::{
     ResourcePair, lower_error_context_to_index, lower_future_to_index, lower_stream_to_index,
diff --git a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
@@ -18,8 +18,8 @@ use core::marker::PhantomData;
 use core::mem::{self, MaybeUninit};
 use core::pin::Pin;
 use core::task::{Context, Poll, Waker, ready};
-use futures::FutureExt;
 use futures::channel::oneshot;
+use futures::{FutureExt as _, stream};
 use std::boxed::Box;
 use std::io::Cursor;
 use std::string::{String, ToString};
@@ -509,26 +509,25 @@ pub trait StreamProducer<D>: Send + 'static {
     ) -> Poll<Result<StreamResult>>;
 }
 
-/// An empty [`StreamProducer`], which will never produce any elements.
-///
-/// [StreamProducer::poll_produce] will always report the stream as dropped.
-#[derive(Copy, Clone, Debug, Hash, Eq, Ord, PartialEq, PartialOrd)]
-pub struct EmptyProducer<T>(PhantomData<fn(T) -> T>);
-
-impl<T> EmptyProducer<T> {
-    /// Constructs a new [EmptyProducer]
-    pub fn new() -> Self {
-        Self::default()
-    }
-}
+impl<T, D> StreamProducer<D> for iter::Empty<T>
+where
+    T: Send + Sync + 'static,
+{
+    type Item = T;
+    type Buffer = Option<Self::Item>;
 
-impl<T> Default for EmptyProducer<T> {
-    fn default() -> Self {
-        Self(PhantomData)
+    fn poll_produce<'a>(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        _: StoreContextMut<'a, D>,
+        _: Destination<'a, Self::Item, Self::Buffer>,
+        _: bool,
+    ) -> Poll<Result<StreamResult>> {
+        Poll::Ready(Ok(StreamResult::Dropped))
     }
 }
 
-impl<T, D> StreamProducer<D> for EmptyProducer<T>
+impl<T, D> StreamProducer<D> for stream::Empty<T>
 where
     T: Send + Sync + 'static,
 {
@@ -640,44 +639,6 @@ impl<D> StreamProducer<D> for bytes::BytesMut {
     }
 }
 
-/// A [`FutureProducer`], which is ready to produce an element.
-#[derive(Copy, Clone, Debug, Hash, Eq, Ord, PartialEq, PartialOrd)]
-pub struct ReadyProducer<T>(Option<T>);
-
-impl<T> ReadyProducer<T> {
-    /// Constructs a new [ReadyProducer].
-    pub fn new(v: T) -> Self {
-        Self(Some(v))
-    }
-}
-
-impl<T> From<T> for ReadyProducer<T> {
-    fn from(v: T) -> Self {
-        Self(Some(v))
-    }
-}
-
-impl<T, D> FutureProducer<D> for ReadyProducer<T>
-where
-    T: Unpin + Send + 'static,
-{
-    type Item = T;
-
-    fn poll_produce(
-        self: Pin<&mut Self>,
-        _: &mut Context<'_>,
-        _: StoreContextMut<D>,
-        _: bool,
-    ) -> Poll<Result<Option<T>>> {
-        let v = self
-            .get_mut()
-            .0
-            .take()
-            .context("polled after returning `Ready`")?;
-        Poll::Ready(Ok(Some(v)))
-    }
-}
-
 /// Represents the buffer for a host- or guest-initiated stream write.
 pub struct Source<'a, T> {
     instance: Instance,
@@ -4234,22 +4195,7 @@ mod tests {
             poll_future_producer(fut.as_mut(), true),
             Poll::Ready(Ok(None)),
         ));
-    }
-
-    #[test]
-    fn ready_producer() {
-        assert!(matches!(
-            poll_future_producer(pin!(ReadyProducer::from(())), false),
-            Poll::Ready(Ok(Some(()))),
-        ));
-        assert!(matches!(
-            poll_future_producer(pin!(ReadyProducer::from(())), true),
-            Poll::Ready(Ok(Some(()))),
-        ));
-    }
 
-    #[test]
-    fn oneshot_producer() {
         let (tx, rx) = oneshot::channel();
         let mut rx = pin!(rx);
         assert!(matches!(
diff --git a/crates/wasmtime/src/runtime/component/mod.rs b/crates/wasmtime/src/runtime/component/mod.rs
@@ -120,9 +120,9 @@ pub use self::component::{Component, ComponentExportIndex};
 #[cfg(feature = "component-model-async")]
 pub use self::concurrent::{
     Access, Accessor, AccessorTask, AsAccessor, Destination, DirectDestination, DirectSource,
-    EmptyProducer, ErrorContext, FutureConsumer, FutureProducer, FutureReader, GuardedFutureReader,
-    GuardedStreamReader, JoinHandle, ReadBuffer, ReadyProducer, Source, StreamConsumer,
-    StreamProducer, StreamReader, StreamResult, VMComponentAsyncStore, VecBuffer, WriteBuffer,
+    ErrorContext, FutureConsumer, FutureProducer, FutureReader, GuardedFutureReader,
+    GuardedStreamReader, JoinHandle, ReadBuffer, Source, StreamConsumer, StreamProducer,
+    StreamReader, StreamResult, VMComponentAsyncStore, VecBuffer, WriteBuffer,
 };
 #[cfg(feature = "component-model-async")]
 pub use self::func::TaskExit;
