feat: add error impl for streams (#4)

* feat: add error impl for stream/future closure This commit adds the implementation for error propagation during stream/future closure. When streams are closed on write, they have the option to send an error context that should be seen by the reader on the next read. Signed-off-by: Victor Adossi <[email protected]> * refactor: facilitate host read returning error context Signed-off-by: Victor Adossi <[email protected]> * fix: pass error context type for component through Signed-off-by: Victor Adossi <[email protected]> * fix: pass through error context type in more places Signed-off-by: Victor Adossi <[email protected]> * chore: improve invalid handle error messages Signed-off-by: Victor Adossi <[email protected]> * wip: clarify comments around global error context management Signed-off-by: Victor Adossi <[email protected]> * wip: finish up edge cases Signed-off-by: Victor Adossi <[email protected]> * chore: remove irrelevant comment Signed-off-by: Victor Adossi <[email protected]> * fix: remove unneeded ref translation Signed-off-by: Victor Adossi <[email protected]> * fix: remove unused err_ctx_ty Signed-off-by: Victor Adossi <[email protected]> * fix: comment regarding writer and error context Co-authored-by: Joel Dice <[email protected]> * fix: remove unused err_ctx_ty future/stream/flat_stream write Signed-off-by: Victor Adossi <[email protected]> --------- Signed-off-by: Victor Adossi <[email protected]> Co-authored-by: Joel Dice <[email protected]>
bytecodealliance · Jan 28, 2025 · 4611d5c · 4611d5c
1 parent d1596df
commit 4611d5c
Show file tree

Hide file tree

Showing 6 changed files with 519 additions and 152 deletions.
diff --git a/crates/cranelift/src/compiler/component.rs b/crates/cranelift/src/compiler/component.rs
@@ -124,13 +124,17 @@ impl<'a> TrampolineCompiler<'a> {
             Trampoline::TaskYield { async_ } => self.translate_task_yield_call(*async_),
             Trampoline::SubtaskDrop { instance } => self.translate_subtask_drop_call(*instance),
             Trampoline::StreamNew { ty } => self.translate_future_or_stream_call(
-                ty.as_u32(),
+                &[ty.as_u32()],
                 None,
                 self.offsets.stream_new(),
                 Vec::new(),
                 ir::types::I64,
             ),
-            Trampoline::StreamRead { ty, options } => {
+            Trampoline::StreamRead {
+                ty,
+                err_ctx_ty,
+                options,
+            } => {
                 if let Some(info) = self.flat_stream_element_info(*ty) {
                     self.translate_flat_stream_call(
                         *ty,
@@ -140,7 +144,7 @@ impl<'a> TrampolineCompiler<'a> {
                     )
                 } else {
                     self.translate_future_or_stream_call(
-                        ty.as_u32(),
+                        &[ty.as_u32(), err_ctx_ty.as_u32()],
                         Some(options),
                         self.offsets.stream_read(),
                         vec![
@@ -162,7 +166,7 @@ impl<'a> TrampolineCompiler<'a> {
                     )
                 } else {
                     self.translate_future_or_stream_call(
-                        ty.as_u32(),
+                        &[ty.as_u32()],
                         Some(options),
                         self.offsets.stream_write(),
                         vec![
@@ -181,31 +185,36 @@ impl<'a> TrampolineCompiler<'a> {
                 self.translate_cancel_call(ty.as_u32(), *async_, self.offsets.stream_cancel_write())
             }
             Trampoline::StreamCloseReadable { ty } => self.translate_future_or_stream_call(
-                ty.as_u32(),
+                &[ty.as_u32()],
                 None,
                 self.offsets.stream_close_readable(),
                 vec![ir::AbiParam::new(ir::types::I32)],
                 ir::types::I8,
             ),
-            Trampoline::StreamCloseWritable { ty } => self.translate_future_or_stream_call(
-                ty.as_u32(),
-                None,
-                self.offsets.stream_close_writable(),
-                vec![
-                    ir::AbiParam::new(ir::types::I32),
-                    ir::AbiParam::new(ir::types::I32),
-                ],
-                ir::types::I8,
-            ),
+            Trampoline::StreamCloseWritable { ty, err_ctx_ty } => self
+                .translate_future_or_stream_call(
+                    &[ty.as_u32(), err_ctx_ty.as_u32()],
+                    None,
+                    self.offsets.stream_close_writable(),
+                    vec![
+                        ir::AbiParam::new(ir::types::I32),
+                        ir::AbiParam::new(ir::types::I32),
+                    ],
+                    ir::types::I8,
+                ),
             Trampoline::FutureNew { ty } => self.translate_future_or_stream_call(
-                ty.as_u32(),
+                &[ty.as_u32()],
                 None,
                 self.offsets.future_new(),
                 Vec::new(),
                 ir::types::I64,
             ),
-            Trampoline::FutureRead { ty, options } => self.translate_future_or_stream_call(
-                ty.as_u32(),
+            Trampoline::FutureRead {
+                ty,
+                err_ctx_ty,
+                options,
+            } => self.translate_future_or_stream_call(
+                &[ty.as_u32(), err_ctx_ty.as_u32()],
                 Some(&options),
                 self.offsets.future_read(),
                 vec![
@@ -215,7 +224,7 @@ impl<'a> TrampolineCompiler<'a> {
                 ir::types::I64,
             ),
             Trampoline::FutureWrite { ty, options } => self.translate_future_or_stream_call(
-                ty.as_u32(),
+                &[ty.as_u32()],
                 Some(options),
                 self.offsets.future_write(),
                 vec![
@@ -231,22 +240,23 @@ impl<'a> TrampolineCompiler<'a> {
                 self.translate_cancel_call(ty.as_u32(), *async_, self.offsets.future_cancel_write())
             }
             Trampoline::FutureCloseReadable { ty } => self.translate_future_or_stream_call(
-                ty.as_u32(),
+                &[ty.as_u32()],
                 None,
                 self.offsets.future_close_readable(),
                 vec![ir::AbiParam::new(ir::types::I32)],
                 ir::types::I8,
             ),
-            Trampoline::FutureCloseWritable { ty } => self.translate_future_or_stream_call(
-                ty.as_u32(),
-                None,
-                self.offsets.future_close_writable(),
-                vec![
-                    ir::AbiParam::new(ir::types::I32),
-                    ir::AbiParam::new(ir::types::I32),
-                ],
-                ir::types::I8,
-            ),
+            Trampoline::FutureCloseWritable { ty, err_ctx_ty } => self
+                .translate_future_or_stream_call(
+                    &[ty.as_u32(), err_ctx_ty.as_u32()],
+                    None,
+                    self.offsets.future_close_writable(),
+                    vec![
+                        ir::AbiParam::new(ir::types::I32),
+                        ir::AbiParam::new(ir::types::I32),
+                    ],
+                    ir::types::I8,
+                ),
             Trampoline::ErrorContextNew { ty, options } => self.translate_error_context_call(
                 *ty,
                 options,
@@ -1109,7 +1119,7 @@ impl<'a> TrampolineCompiler<'a> {
 
     fn translate_future_or_stream_call(
         &mut self,
-        ty: u32,
+        tys: &[u32],
         options: Option<&CanonicalOptions>,
         offset: u32,
         params: Vec<ir::AbiParam>,
@@ -1165,7 +1175,10 @@ impl<'a> TrampolineCompiler<'a> {
         }
 
         host_sig.params.push(ir::AbiParam::new(ir::types::I32));
-        host_args.push(self.builder.ins().iconst(ir::types::I32, i64::from(ty)));
+
+        for ty in tys {
+            host_args.push(self.builder.ins().iconst(ir::types::I32, i64::from(*ty)));
+        }
 
         host_sig.params.extend(params);
         host_args.extend(args[2..].iter().copied());

diff --git a/crates/environ/src/component/dfg.rs b/crates/environ/src/component/dfg.rs
@@ -308,6 +308,7 @@ pub enum Trampoline {
     },
     StreamRead {
         ty: TypeStreamTableIndex,
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
         options: CanonicalOptions,
     },
     StreamWrite {
@@ -327,12 +328,14 @@ pub enum Trampoline {
     },
     StreamCloseWritable {
         ty: TypeStreamTableIndex,
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
     },
     FutureNew {
         ty: TypeFutureTableIndex,
     },
     FutureRead {
         ty: TypeFutureTableIndex,
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
         options: CanonicalOptions,
     },
     FutureWrite {
@@ -352,6 +355,7 @@ pub enum Trampoline {
     },
     FutureCloseWritable {
         ty: TypeFutureTableIndex,
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
     },
     ErrorContextNew {
         ty: TypeComponentLocalErrorContextTableIndex,
@@ -789,8 +793,13 @@ impl LinearizeDfg<'_> {
                 instance: *instance,
             },
             Trampoline::StreamNew { ty } => info::Trampoline::StreamNew { ty: *ty },
-            Trampoline::StreamRead { ty, options } => info::Trampoline::StreamRead {
+            Trampoline::StreamRead {
+                ty,
+                err_ctx_ty,
+                options,
+            } => info::Trampoline::StreamRead {
                 ty: *ty,
+                err_ctx_ty: *err_ctx_ty,
                 options: self.options(options),
             },
             Trampoline::StreamWrite { ty, options } => info::Trampoline::StreamWrite {
@@ -808,12 +817,20 @@ impl LinearizeDfg<'_> {
             Trampoline::StreamCloseReadable { ty } => {
                 info::Trampoline::StreamCloseReadable { ty: *ty }
             }
-            Trampoline::StreamCloseWritable { ty } => {
-                info::Trampoline::StreamCloseWritable { ty: *ty }
+            Trampoline::StreamCloseWritable { ty, err_ctx_ty } => {
+                info::Trampoline::StreamCloseWritable {
+                    ty: *ty,
+                    err_ctx_ty: *err_ctx_ty,
+                }
             }
             Trampoline::FutureNew { ty } => info::Trampoline::FutureNew { ty: *ty },
-            Trampoline::FutureRead { ty, options } => info::Trampoline::FutureRead {
+            Trampoline::FutureRead {
+                ty,
+                err_ctx_ty,
+                options,
+            } => info::Trampoline::FutureRead {
                 ty: *ty,
+                err_ctx_ty: *err_ctx_ty,
                 options: self.options(options),
             },
             Trampoline::FutureWrite { ty, options } => info::Trampoline::FutureWrite {
@@ -831,8 +848,11 @@ impl LinearizeDfg<'_> {
             Trampoline::FutureCloseReadable { ty } => {
                 info::Trampoline::FutureCloseReadable { ty: *ty }
             }
-            Trampoline::FutureCloseWritable { ty } => {
-                info::Trampoline::FutureCloseWritable { ty: *ty }
+            Trampoline::FutureCloseWritable { ty, err_ctx_ty } => {
+                info::Trampoline::FutureCloseWritable {
+                    ty: *ty,
+                    err_ctx_ty: *err_ctx_ty,
+                }
             }
             Trampoline::ErrorContextNew { ty, options } => info::Trampoline::ErrorContextNew {
                 ty: *ty,

diff --git a/crates/environ/src/component/info.rs b/crates/environ/src/component/info.rs
@@ -738,6 +738,10 @@ pub enum Trampoline {
     StreamRead {
         /// The table index for the specific `stream` type and caller instance.
         ty: TypeStreamTableIndex,
+
+        /// The table index for the `error-context` type in the caller instance.
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
+
         /// Any options (e.g. string encoding) to use when storing values to
         /// memory.
         options: CanonicalOptions,
@@ -784,6 +788,9 @@ pub enum Trampoline {
     StreamCloseWritable {
         /// The table index for the specific `stream` type and caller instance.
         ty: TypeStreamTableIndex,
+
+        /// The table index for the `error-context` type in the caller instance.
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
     },
 
     /// A `future.new` intrinsic to create a new `future` handle of the
@@ -797,6 +804,10 @@ pub enum Trampoline {
     FutureRead {
         /// The table index for the specific `future` type and caller instance.
         ty: TypeFutureTableIndex,
+
+        /// The table index for the `error-context` type in the caller instance.
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
+
         /// Any options (e.g. string encoding) to use when storing values to
         /// memory.
         options: CanonicalOptions,
@@ -843,6 +854,9 @@ pub enum Trampoline {
     FutureCloseWritable {
         /// The table index for the specific `future` type and caller instance.
         ty: TypeFutureTableIndex,
+
+        /// The table index for the `error-context` type in the caller instance.
+        err_ctx_ty: TypeComponentLocalErrorContextTableIndex,
     },
 
     /// A `error-context.new` intrinsic to create a new `error-context` with a

diff --git a/crates/environ/src/component/translate/inline.rs b/crates/environ/src/component/translate/inline.rs
@@ -754,12 +754,17 @@ impl<'a> Inliner<'a> {
                 else {
                     unreachable!()
                 };
+                let err_ctx_ty = types.error_context_table_type()?;
                 let options = self.adapter_options(frame, types, options);
                 let options = self.canonical_options(options);
-                let index = self
-                    .result
-                    .trampolines
-                    .push((*func, dfg::Trampoline::StreamRead { ty, options }));
+                let index = self.result.trampolines.push((
+                    *func,
+                    dfg::Trampoline::StreamRead {
+                        ty,
+                        err_ctx_ty,
+                        options,
+                    },
+                ));
                 frame.funcs.push(dfg::CoreDef::Trampoline(index));
             }
             StreamWrite { ty, func, options } => {
@@ -824,10 +829,11 @@ impl<'a> Inliner<'a> {
                 else {
                     unreachable!()
                 };
-                let index = self
-                    .result
-                    .trampolines
-                    .push((*func, dfg::Trampoline::StreamCloseWritable { ty }));
+                let err_ctx_ty = types.error_context_table_type()?;
+                let index = self.result.trampolines.push((
+                    *func,
+                    dfg::Trampoline::StreamCloseWritable { ty, err_ctx_ty },
+                ));
                 frame.funcs.push(dfg::CoreDef::Trampoline(index));
             }
             FutureNew { ty, func } => {
@@ -848,12 +854,17 @@ impl<'a> Inliner<'a> {
                 else {
                     unreachable!()
                 };
+                let err_ctx_ty = types.error_context_table_type()?;
                 let options = self.adapter_options(frame, types, options);
                 let options = self.canonical_options(options);
-                let index = self
-                    .result
-                    .trampolines
-                    .push((*func, dfg::Trampoline::FutureRead { ty, options }));
+                let index = self.result.trampolines.push((
+                    *func,
+                    dfg::Trampoline::FutureRead {
+                        ty,
+                        err_ctx_ty,
+                        options,
+                    },
+                ));
                 frame.funcs.push(dfg::CoreDef::Trampoline(index));
             }
             FutureWrite { ty, func, options } => {
@@ -918,10 +929,11 @@ impl<'a> Inliner<'a> {
                 else {
                     unreachable!()
                 };
-                let index = self
-                    .result
-                    .trampolines
-                    .push((*func, dfg::Trampoline::FutureCloseWritable { ty }));
+                let err_ctx_ty = types.error_context_table_type()?;
+                let index = self.result.trampolines.push((
+                    *func,
+                    dfg::Trampoline::FutureCloseWritable { ty, err_ctx_ty },
+                ));
                 frame.funcs.push(dfg::CoreDef::Trampoline(index));
             }
             ErrorContextNew { func, options } => {