wasmtime/runtime/component/concurrent/
future_stream_any.rs

1//! Implementation of [`FutureAny`] and [`StreamAny`].
2
3use crate::component::concurrent::futures_and_streams::{self, TransmitOrigin};
4use crate::component::func::{LiftContext, LowerContext, desc};
5use crate::component::matching::InstanceType;
6use crate::component::resources::HostResourceIndex;
7use crate::component::types::{FutureType, StreamType};
8use crate::component::{
9    ComponentInstanceId, ComponentType, FutureReader, Lift, Lower, StreamReader,
10};
11use crate::store::StoreOpaque;
12use crate::{AsContextMut, Result, bail};
13use core::any::TypeId;
14use core::mem::MaybeUninit;
15use wasmtime_environ::component::{
16    CanonicalAbiInfo, InterfaceType, TypeFutureTableIndex, TypeStreamTableIndex,
17};
18
19/// Represents a type-erased component model `future`.
20///
21/// This type is similar to [`ResourceAny`](crate::component::ResourceAny)
22/// where it's a static guarantee that it represents a component model
23/// `future`, but it does not contain any information about the underlying type
24/// that is associated with this future. This is intended to be used in
25/// "dynamically typed" situations where embedders may not know ahead of time
26/// the type of a `future` being used by component that is loaded.
27///
28/// # Closing futures
29///
30/// A [`FutureAny`] represents a resource that is owned by a [`Store`]. Proper
31/// disposal of a future requires invoking the [`FutureAny::close`] method to
32/// ensure that this handle does not leak. If [`FutureAny::close`] is not
33/// called then memory will not be leaked once the owning [`Store`] is dropped,
34/// but the resource handle will be leaked until the [`Store`] is dropped.
35///
36/// Note that special care should be taken when handling `FutureAny` in async
37/// functions. In this scenario an async host function can be cancelled
38/// (dropped) at any time and if a `FutureAny` it holds isn't closed then the
39/// future will be leaked within the store until the store is dropped.
40///
41/// # Cloning
42///
43/// Cloning a [`FutureAny`] does not duplicate the underlying future: all clones
44/// refer to the same read end, whose ownership can only be transferred once.
45/// After one clone has been lowered into a guest, piped, or closed, attempting
46/// any of those with another clone returns an error.
47///
48/// [`Store`]: crate::Store
49#[derive(Debug, Clone, PartialEq)]
50pub struct FutureAny {
51    idx: HostResourceIndex,
52    ty: PayloadType<FutureType>,
53}
54
55impl FutureAny {
56    /// Attempts to convert this [`FutureAny`] to a [`FutureReader<T>`]
57    /// with a statically known type.
58    ///
59    /// # Errors
60    ///
61    /// This function will return an error if `T` does not match the type of
62    /// value on this future.
63    pub fn try_into_future_reader<T>(self) -> Result<FutureReader<T>>
64    where
65        T: ComponentType + 'static,
66    {
67        self.ty
68            .typecheck_host::<T>(FutureType::equivalent_payload_host::<T>)?;
69        Ok(FutureReader::new_(self.idx))
70    }
71
72    /// Attempts to convert `reader` to a [`FutureAny`], erasing its statically
73    /// known type.
74    ///
75    /// # Errors
76    ///
77    /// This function will return an error if `reader` does not belong to
78    /// `store`.
79    pub fn try_from_future_reader<T>(
80        mut store: impl AsContextMut,
81        reader: FutureReader<T>,
82    ) -> Result<Self>
83    where
84        T: ComponentType + 'static,
85    {
86        let store = store.as_context_mut();
87        let ty = match store.0.transmit_origin(reader.idx())? {
88            TransmitOrigin::Host { .. } => PayloadType::new_host::<T>(),
89            TransmitOrigin::GuestFuture(id, ty) => PayloadType::new_guest_future(store.0, id, ty),
90            TransmitOrigin::GuestStream(..) => bail!("not a future"),
91        };
92        Ok(FutureAny {
93            idx: reader.idx(),
94            ty,
95        })
96    }
97
98    fn lift_from_index(cx: &mut LiftContext<'_>, ty: InterfaceType, index: u32) -> Result<Self> {
99        let idx = futures_and_streams::lift_index_to_future(cx, ty, index)?;
100        let InterfaceType::Future(ty) = ty else {
101            unreachable!()
102        };
103        let ty = cx.types[ty].ty;
104        Ok(FutureAny {
105            idx,
106            // Note that this future might actually be a host-originating
107            // future which means that this ascription of "the type is the
108            // guest" may be slightly in accurate. The guest, however, has the
109            // most accurate view of what type this future has so that should
110            // be reasonable to ascribe as the type here regardless.
111            ty: PayloadType::Guest(FutureType::from(ty, &cx.instance_type())),
112        })
113    }
114
115    /// Close this `FutureAny`.
116    ///
117    /// This will close this future and cause any write that happens later to
118    /// returned `DROPPED`.
119    ///
120    /// # Errors
121    ///
122    /// Returns an error if this future has already been closed or otherwise
123    /// transferred (e.g. lowered into a guest), or if it does not belong to
124    /// `store`.
125    pub fn close(&mut self, mut store: impl AsContextMut) -> Result<()> {
126        futures_and_streams::future_close(store.as_context_mut().0, self.idx)
127    }
128}
129
130unsafe impl ComponentType for FutureAny {
131    const ABI: CanonicalAbiInfo = CanonicalAbiInfo::SCALAR4;
132
133    type Lower = <u32 as ComponentType>::Lower;
134
135    fn typecheck(ty: &InterfaceType, _types: &InstanceType<'_>) -> Result<()> {
136        match ty {
137            InterfaceType::Future(_) => Ok(()),
138            other => bail!("expected `future`, found `{}`", desc(other)),
139        }
140    }
141}
142
143unsafe impl Lower for FutureAny {
144    fn linear_lower_to_flat<T>(
145        &self,
146        cx: &mut LowerContext<'_, T>,
147        ty: InterfaceType,
148        dst: &mut MaybeUninit<Self::Lower>,
149    ) -> Result<()> {
150        futures_and_streams::lower_future_to_index(self.idx, cx, ty)?.linear_lower_to_flat(
151            cx,
152            InterfaceType::U32,
153            dst,
154        )
155    }
156
157    fn linear_lower_to_memory<T>(
158        &self,
159        cx: &mut LowerContext<'_, T>,
160        ty: InterfaceType,
161        offset: usize,
162    ) -> Result<()> {
163        futures_and_streams::lower_future_to_index(self.idx, cx, ty)?.linear_lower_to_memory(
164            cx,
165            InterfaceType::U32,
166            offset,
167        )
168    }
169}
170
171unsafe impl Lift for FutureAny {
172    fn linear_lift_from_flat(
173        cx: &mut LiftContext<'_>,
174        ty: InterfaceType,
175        src: &Self::Lower,
176    ) -> Result<Self> {
177        let index = u32::linear_lift_from_flat(cx, InterfaceType::U32, src)?;
178        Self::lift_from_index(cx, ty, index)
179    }
180
181    fn linear_lift_from_memory(
182        cx: &mut LiftContext<'_>,
183        ty: InterfaceType,
184        bytes: &[u8],
185    ) -> Result<Self> {
186        let index = u32::linear_lift_from_memory(cx, InterfaceType::U32, bytes)?;
187        Self::lift_from_index(cx, ty, index)
188    }
189}
190
191/// Represents a type-erased component model `stream`.
192///
193/// This type is similar to [`ResourceAny`](crate::component::ResourceAny)
194/// where it's a static guarantee that it represents a component model
195/// `stream`, but it does not contain any information about the underlying type
196/// that is associated with this stream. This is intended to be used in
197/// "dynamically typed" situations where embedders may not know ahead of time
198/// the type of a `stream` being used by component that is loaded.
199///
200/// # Closing streams
201///
202/// A [`StreamAny`] represents a resource that is owned by a [`Store`]. Proper
203/// disposal of a stream requires invoking the [`StreamAny::close`] method to
204/// ensure that this handle does not leak. If [`StreamAny::close`] is not
205/// called then memory will not be leaked once the owning [`Store`] is dropped,
206/// but the resource handle will be leaked until the [`Store`] is dropped.
207///
208/// Note that special care should be taken when handling `StreamAny` in async
209/// functions. In this scenario an async host function can be cancelled
210/// (dropped) at any time and if a `StreamAny` it holds isn't closed then the
211/// stream will be leaked within the store until the store is dropped.
212///
213/// # Cloning
214///
215/// Cloning a [`StreamAny`] does not duplicate the underlying stream: all clones
216/// refer to the same read end, whose ownership can only be transferred once.
217/// After one clone has been lowered into a guest, piped, or closed, attempting
218/// any of those with another clone returns an error.
219///
220/// [`Store`]: crate::Store
221#[derive(Debug, Clone, PartialEq)]
222pub struct StreamAny {
223    idx: HostResourceIndex,
224    ty: PayloadType<StreamType>,
225}
226
227impl StreamAny {
228    /// Attempts to convert this [`StreamAny`] to a [`StreamReader<T>`]
229    /// with a statically known type.
230    ///
231    /// # Errors
232    ///
233    /// This function will return an error if `T` does not match the type of
234    /// value on this stream.
235    pub fn try_into_stream_reader<T>(self) -> Result<StreamReader<T>>
236    where
237        T: ComponentType + 'static,
238    {
239        self.ty
240            .typecheck_host::<T>(StreamType::equivalent_payload_host::<T>)?;
241        Ok(StreamReader::new_(self.idx))
242    }
243
244    /// Attempts to convert `reader` to a [`StreamAny`], erasing its statically
245    /// known type.
246    ///
247    /// # Errors
248    ///
249    /// This function will return an error if `reader` does not belong to
250    /// `store`.
251    pub fn try_from_stream_reader<T>(
252        mut store: impl AsContextMut,
253        reader: StreamReader<T>,
254    ) -> Result<Self>
255    where
256        T: ComponentType + 'static,
257    {
258        let store = store.as_context_mut();
259        let ty = match store.0.transmit_origin(reader.idx())? {
260            TransmitOrigin::Host { .. } => PayloadType::new_host::<T>(),
261            TransmitOrigin::GuestStream(id, ty) => PayloadType::new_guest_stream(store.0, id, ty),
262            TransmitOrigin::GuestFuture(..) => bail!("not a stream"),
263        };
264        Ok(StreamAny {
265            idx: reader.idx(),
266            ty,
267        })
268    }
269
270    fn lift_from_index(cx: &mut LiftContext<'_>, ty: InterfaceType, index: u32) -> Result<Self> {
271        let idx = futures_and_streams::lift_index_to_stream(cx, ty, index)?;
272        let InterfaceType::Stream(ty) = ty else {
273            unreachable!()
274        };
275        let ty = cx.types[ty].ty;
276        Ok(StreamAny {
277            idx,
278            // Note that this stream might actually be a host-originating, but
279            // see the documentation in `FutureAny::lift_from_index` for why
280            // this should be ok.
281            ty: PayloadType::Guest(StreamType::from(ty, &cx.instance_type())),
282        })
283    }
284
285    /// Close this `StreamAny`.
286    ///
287    /// This will close this stream and cause any write that happens later to
288    /// returned `DROPPED`.
289    ///
290    /// # Errors
291    ///
292    /// Returns an error if this stream has already been closed or otherwise
293    /// transferred (e.g. lowered into a guest), or if it does not belong to
294    /// `store`.
295    pub fn close(&mut self, mut store: impl AsContextMut) -> Result<()> {
296        futures_and_streams::stream_close(store.as_context_mut().0, self.idx)
297    }
298}
299
300unsafe impl ComponentType for StreamAny {
301    const ABI: CanonicalAbiInfo = CanonicalAbiInfo::SCALAR4;
302
303    type Lower = <u32 as ComponentType>::Lower;
304
305    fn typecheck(ty: &InterfaceType, _types: &InstanceType<'_>) -> Result<()> {
306        match ty {
307            InterfaceType::Stream(_) => Ok(()),
308            other => bail!("expected `stream`, found `{}`", desc(other)),
309        }
310    }
311}
312
313unsafe impl Lower for StreamAny {
314    fn linear_lower_to_flat<T>(
315        &self,
316        cx: &mut LowerContext<'_, T>,
317        ty: InterfaceType,
318        dst: &mut MaybeUninit<Self::Lower>,
319    ) -> Result<()> {
320        futures_and_streams::lower_stream_to_index(self.idx, cx, ty)?.linear_lower_to_flat(
321            cx,
322            InterfaceType::U32,
323            dst,
324        )
325    }
326
327    fn linear_lower_to_memory<T>(
328        &self,
329        cx: &mut LowerContext<'_, T>,
330        ty: InterfaceType,
331        offset: usize,
332    ) -> Result<()> {
333        futures_and_streams::lower_stream_to_index(self.idx, cx, ty)?.linear_lower_to_memory(
334            cx,
335            InterfaceType::U32,
336            offset,
337        )
338    }
339}
340
341unsafe impl Lift for StreamAny {
342    fn linear_lift_from_flat(
343        cx: &mut LiftContext<'_>,
344        ty: InterfaceType,
345        src: &Self::Lower,
346    ) -> Result<Self> {
347        let index = u32::linear_lift_from_flat(cx, InterfaceType::U32, src)?;
348        Self::lift_from_index(cx, ty, index)
349    }
350
351    fn linear_lift_from_memory(
352        cx: &mut LiftContext<'_>,
353        ty: InterfaceType,
354        bytes: &[u8],
355    ) -> Result<Self> {
356        let index = u32::linear_lift_from_memory(cx, InterfaceType::U32, bytes)?;
357        Self::lift_from_index(cx, ty, index)
358    }
359}
360
361#[derive(Debug, Clone, PartialEq)]
362enum PayloadType<T> {
363    Guest(T),
364    Host(TypeId),
365}
366
367impl PayloadType<FutureType> {
368    fn new_guest_future(
369        store: &StoreOpaque,
370        id: ComponentInstanceId,
371        ty: TypeFutureTableIndex,
372    ) -> Self {
373        let types = InstanceType::new(&store.component_instance(id));
374        let ty = types.types[ty].ty;
375        PayloadType::Guest(FutureType::from(ty, &types))
376    }
377}
378
379impl PayloadType<StreamType> {
380    fn new_guest_stream(
381        store: &StoreOpaque,
382        id: ComponentInstanceId,
383        ty: TypeStreamTableIndex,
384    ) -> Self {
385        let types = InstanceType::new(&store.component_instance(id));
386        let ty = types.types[ty].ty;
387        PayloadType::Guest(StreamType::from(ty, &types))
388    }
389}
390
391impl<T> PayloadType<T> {
392    fn new_host<P>() -> Self
393    where
394        P: ComponentType + 'static,
395    {
396        PayloadType::Host(TypeId::of::<P>())
397    }
398
399    fn typecheck_host<P>(&self, equivalent: fn(&T) -> Result<()>) -> Result<()>
400    where
401        P: ComponentType + 'static,
402    {
403        match self {
404            Self::Guest(ty) => equivalent(ty),
405            Self::Host(id) => {
406                if *id == TypeId::of::<P>() {
407                    Ok(())
408                } else {
409                    bail!("future payload types differ")
410                }
411            }
412        }
413    }
414}
wasmtime/runtime/component/concurrent/future_stream_any.rs

wasmtime/runtime/component/concurrent/
future_stream_any.rs
