Repository: bytecodealliance/wasmtime
Pull request #7461 (2023-11-02 17:33:51)

Remove a writer task in wasi-http
This commit updates the wasi-http `BodyWriteStream` implementation to avoid an extra task spawned to manage the outgoing body. Some local profiling shows that a "hello world" component currently maxes out at 500rps locally which is suprisingly low.

Some investigation shows that one of the primary reasons for this is concurrency-related issues. For example `taskset -c 1 wasmtime serve ...` has a significantly better rps than without `taskset`. My hypothesis is that the extra tasks for both reading the incoming body and writing the outgoing body are introducing opportunities for the guest to block and get work-stolen to another thread in the "happy path" where reads/writes are immediately ready and don't need blocking.

For example writing to the body currently sends the chunk of bytes to a task which then sends it to another task. This means that if a guest writes some bytes and then waits for readiness it'll synchronize on these bytes by doing a round-trip with another task. This opportunity for blocking apparently lets something, be it Linux or the Tokio scheduler, to work-steal the wasm task to another thread.

Once a wasm task is work-stolen to another thread it can significantly increase the cost of VM management syscalls because any happy path the kernel might have for a single thread is no longer applicable, meaning that many costs start shooting through the roof.

The fix here in this PR is to remove a helper task when writing HTTP bodies. This extra task didn't end up doing much and was largely just a proxy into a channel, so the channel is now directly manipulated instead. Namely the `flush` implementation in the old worker task didn't actually do anything and so the only thing that needs to be managed in the stream is sending chunks along through the channel. This proved relatively easy and enables the "happy path" where if a body is written and then flushed it's immediately "ready" so no opportunity for a context switch happens and wasm stays planted on its thread.

The end result of this PR is that `wasmtime serve` goes from 500rps to 5000rps (10x increase) for the simple "hello world" example that I'm working with.

<!--
Please make sure you include the following information:

- If this work has been discussed elsewhere, please include a link to that
  conversation. If it was discussed in an issue, just mention "issue #...".

- Explain why this change is needed. If the details are in an issue already,
  this can be brief.

Our development process is documented in the Wasmtime book:
https://docs.wasmtime.dev/contributing-development-process.html

Please ensure all communication follows the code of conduct:
https://github.com/bytecodealliance/wasmtime/blob/main/CODE_OF_CONDUCT.md
-->


alexcrichton: Good point @pchickey, how about this amended approach? I had to increase the channel capacity by 1 to get a test to pass but otherwise this seems like a good way to avoid that. This way we don't allow writes unless there's guaranteed capacity in the channel so we're guaranteed channel sends always succeed.

Patch:
diff --git a/cranelift/codegen/src/isa/riscv64/inst.isle b/cranelift/codegen/src/isa/riscv64/inst.isle
index 0c4e1e3b5..18ddcfc54 100644
--- a/cranelift/codegen/src/isa/riscv64/inst.isle
+++ b/cranelift/codegen/src/isa/riscv64/inst.isle
@@ -1885,6 +1885,14 @@
 (decl imm12_from_value (Imm12) Value)
 (extractor (imm12_from_value n) (i64_from_iconst (imm12_from_i64 n)))
 
+;; Conceptually the same as `imm12_from_value`, but tries negating the constant
+;; value (first sign-extending to handle narrow widths).
+(decl pure partial imm12_from_negated_value (Value) Imm12)
+(rule
+  (imm12_from_negated_value (has_type ty (iconst n)))
+  (if-let (imm12_from_u64 imm) (i64_as_u64 (i64_neg (i64_sextend_imm64 ty n))))
+  imm)
+
 (decl imm12_from_u64 (Imm12) u64)
 (extern extractor imm12_from_u64 imm12_from_u64)
 
diff --git a/cranelift/codegen/src/isa/riscv64/inst/emit.rs b/cranelift/codegen/src/isa/riscv64/inst/emit.rs
index 47e5b8a15..fb63e30a0 100644
--- a/cranelift/codegen/src/isa/riscv64/inst/emit.rs
+++ b/cranelift/codegen/src/isa/riscv64/inst/emit.rs
@@ -377,25 +377,42 @@ impl Inst {
                     alu_op @ (AluOPRRR::And
                     | AluOPRRR::Or
                     | AluOPRRR::Xor
-                    | AluOPRRR::Sub
                     | AluOPRRR::Addw
-                    | AluOPRRR::Subw
                     | AluOPRRR::Mul),
                 rd,
                 rs1,
                 rs2,
-            } if rd.to_reg() == rs1 && reg_is_compressible(rs1) && reg_is_compressible(rs2) => {
+            } if (rd.to_reg() == rs1 || rd.to_reg() == rs2)
+                && reg_is_compressible(rs1)
+                && reg_is_compressible(rs2) =>
+            {
                 let op = match alu_op {
                     AluOPRRR::And => CaOp::CAnd,
                     AluOPRRR::Or => CaOp::COr,
                     AluOPRRR::Xor => CaOp::CXor,
-                    AluOPRRR::Sub => CaOp::CSub,
                     AluOPRRR::Addw => CaOp::CAddw,
-                    AluOPRRR::Subw => CaOp::CSubw,
                     AluOPRRR::Mul if has_zcb && has_m => CaOp::CMul,
                     _ => return None,
                 };
+                // The canonical expansion for these instruction has `rd == rs1`, but
+                // these are all comutative operations, so we can swap the operands.
+                let src = if rd.to_reg() == rs1 { rs2 } else { rs1 };
 
+                sink.put2(encode_ca_type(op, rd, src));
+            }
+
+            // The sub instructions are non comutative, so we can't swap the operands.
+            Inst::AluRRR {
+                alu_op: alu_op @ (AluOPRRR::Sub | AluOPRRR::Subw),
+                rd,
+                rs1,
+                rs2,
+            } if rd.to_reg() == rs1 && reg_is_compressible(rs1) && reg_is_compressible(rs2) => {
+                let op = match alu_op {
+                    AluOPRRR::Sub => CaOp::CSub,
+                    AluOPRRR::Subw => CaOp::CSubw,
+                    _ => return None,
+                };
                 sink.put2(encode_ca_type(op, rd, rs2));
             }
 
diff --git a/cranelift/codegen/src/isa/riscv64/lower.isle b/cranelift/codegen/src/isa/riscv64/lower.isle
index e3fa6a9f2..059060476 100644
--- a/cranelift/codegen/src/isa/riscv64/lower.isle
+++ b/cranelift/codegen/src/isa/riscv64/lower.isle
@@ -324,107 +324,116 @@
 (rule 2 (lower (has_type $I128 (isub x y)))
   (i128_sub x y))
 
+;; Switch to an `addi` by a negative if we can fit the value in an `imm12`.
+(rule 3 (lower (has_type (ty_int_ref_scalar_64 ty) (isub x y)))
+  (if-let imm12_neg (imm12_from_negated_value y))
+  (alu_rr_imm12 (select_addi ty) x imm12_neg))
+
+(rule 4 (lower (has_type (ty_int_ref_scalar_64 ty) (isub x y)))
+  (if-let imm12_neg (imm12_from_negated_value x))
+  (alu_rr_imm12 (select_addi ty) y imm12_neg))
+
 ;; SIMD Vectors
-(rule 3 (lower (has_type (ty_vec_fits_in_register ty) (isub x y)))
+(rule 5 (lower (has_type (ty_vec_fits_in_register ty) (isub x y)))
   (rv_vsub_vv x y (unmasked) ty))
 
-(rule 4 (lower (has_type (ty_vec_fits_in_register ty) (isub x (splat y))))
+(rule 6 (lower (has_type (ty_vec_fits_in_register ty) (isub x (splat y))))
   (rv_vsub_vx x y (unmasked) ty))
 
-(rule 5 (lower (has_type (ty_vec_fits_in_register ty) (isub x (splat (sextend y @ (value_type sext_ty))))))
+(rule 7 (lower (has_type (ty_vec_fits_in_register ty) (isub x (splat (sextend y @ (value_type sext_ty))))))
   (if-let half_ty (ty_half_width ty))
   (if-let $true (ty_equal (lane_type half_ty) sext_ty))
   (rv_vwsub_wx x y (unmasked) (vstate_mf2 half_ty)))
 
-(rule 5 (lower (has_type (ty_vec_fits_in_register ty) (isub x (splat (uextend y @ (value_type uext_ty))))))
+(rule 7 (lower (has_type (ty_vec_fits_in_register ty) (isub x (splat (uextend y @ (value_type uext_ty))))))
   (if-let half_ty (ty_half_width ty))
   (if-let $true (ty_equal (lane_type half_ty) uext_ty))
   (rv_vwsubu_wx x y (unmasked) (vstate_mf2 half_ty)))
 
-(rule 6 (lower (has_type (ty_vec_fits_in_register ty) (isub (splat x) y)))
+(rule 8 (lower (has_type (ty_vec_fits_in_register ty) (isub (splat x) y)))
   (rv_vrsub_vx y x (unmasked) ty))
 
-(rule 7 (lower (has_type (ty_vec_fits_in_register ty) (isub x y)))
+(rule 9 (lower (has_type (ty_vec_fits_in_register ty) (isub x y)))
   (if-let x_imm (replicated_imm5 x))
   (rv_vrsub_vi y x_imm (unmasked) ty))
 
 
 ;; Signed Widening Low Subtractions
 
-(rule 5 (lower (has_type (ty_vec_fits_in_register _) (isub x (swiden_low y @ (value_type in_ty)))))
+(rule 7 (lower (has_type (ty_vec_fits_in_register _) (isub x (swiden_low y @ (value_type in_ty)))))
   (rv_vwsub_wv x y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_low x @ (value_type in_ty))
-                                                           (swiden_low y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_low x @ (value_type in_ty))
+                                                            (swiden_low y))))
   (rv_vwsub_vv x y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_low x @ (value_type in_ty))
-                                                           (splat (sextend y @ (value_type sext_ty))))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_low x @ (value_type in_ty))
+                                                            (splat (sextend y @ (value_type sext_ty))))))
   (if-let $true (ty_equal (lane_type in_ty) sext_ty))
   (rv_vwsub_vx x y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
 ;; Signed Widening High Subtractions
 ;; These are the same as the low widenings, but we first slide down the inputs.
 
-(rule 5 (lower (has_type (ty_vec_fits_in_register _) (isub x (swiden_high y @ (value_type in_ty)))))
+(rule 7 (lower (has_type (ty_vec_fits_in_register _) (isub x (swiden_high y @ (value_type in_ty)))))
   (rv_vwsub_wv x (gen_slidedown_half in_ty y) (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_high x @ (value_type in_ty))
-                                                           (swiden_high y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_high x @ (value_type in_ty))
+                                                            (swiden_high y))))
   (rv_vwsub_vv (gen_slidedown_half in_ty x) (gen_slidedown_half in_ty y) (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_high x @ (value_type in_ty))
-                                                           (splat (sextend y @ (value_type sext_ty))))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_high x @ (value_type in_ty))
+                                                            (splat (sextend y @ (value_type sext_ty))))))
   (if-let $true (ty_equal (lane_type in_ty) sext_ty))
   (rv_vwsub_vx (gen_slidedown_half in_ty x) y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
 ;; Unsigned Widening Low Subtractions
 
-(rule 5 (lower (has_type (ty_vec_fits_in_register _) (isub x (uwiden_low y @ (value_type in_ty)))))
+(rule 7 (lower (has_type (ty_vec_fits_in_register _) (isub x (uwiden_low y @ (value_type in_ty)))))
   (rv_vwsubu_wv x y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_low x @ (value_type in_ty))
-                                                           (uwiden_low y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_low x @ (value_type in_ty))
+                                                            (uwiden_low y))))
   (rv_vwsubu_vv x y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_low x @ (value_type in_ty))
-                                                           (splat (uextend y @ (value_type uext_ty))))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_low x @ (value_type in_ty))
+                                                            (splat (uextend y @ (value_type uext_ty))))))
   (if-let $true (ty_equal (lane_type in_ty) uext_ty))
   (rv_vwsubu_vx x y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
 ;; Unsigned Widening High Subtractions
 ;; These are the same as the low widenings, but we first slide down the inputs.
 
-(rule 5 (lower (has_type (ty_vec_fits_in_register _) (isub x (uwiden_high y @ (value_type in_ty)))))
+(rule 7 (lower (has_type (ty_vec_fits_in_register _) (isub x (uwiden_high y @ (value_type in_ty)))))
   (rv_vwsubu_wv x (gen_slidedown_half in_ty y) (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_high x @ (value_type in_ty))
-                                                           (uwiden_high y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_high x @ (value_type in_ty))
+                                                            (uwiden_high y))))
   (rv_vwsubu_vv (gen_slidedown_half in_ty x) (gen_slidedown_half in_ty y) (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_high x @ (value_type in_ty))
-                                                           (splat (uextend y @ (value_type uext_ty))))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_high x @ (value_type in_ty))
+                                                            (splat (uextend y @ (value_type uext_ty))))))
   (if-let $true (ty_equal (lane_type in_ty) uext_ty))
   (rv_vwsubu_vx (gen_slidedown_half in_ty x) y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
 ;; Signed Widening Mixed High/Low Subtractions
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_low x @ (value_type in_ty))
-                                                           (swiden_high y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_low x @ (value_type in_ty))
+                                                            (swiden_high y))))
   (rv_vwsub_vv x (gen_slidedown_half in_ty y) (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_high x @ (value_type in_ty))
-                                                           (swiden_low y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (swiden_high x @ (value_type in_ty))
+                                                            (swiden_low y))))
   (rv_vwsub_vv (gen_slidedown_half in_ty x) y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
 ;; Unsigned Widening Mixed High/Low Subtractions
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_low x @ (value_type in_ty))
-                                                           (uwiden_high y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_low x @ (value_type in_ty))
+                                                            (uwiden_high y))))
   (rv_vwsubu_vv x (gen_slidedown_half in_ty y) (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
-(rule 8 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_high x @ (value_type in_ty))
-                                                           (uwiden_low y))))
+(rule 10 (lower (has_type (ty_vec_fits_in_register _) (isub (uwiden_high x @ (value_type in_ty))
+                                                            (uwiden_low y))))
   (rv_vwsubu_vv (gen_slidedown_half in_ty x) y (unmasked) (vstate_mf2 (ty_half_lanes in_ty))))
 
 
diff --git a/crates/wasi-http/src/body.rs b/crates/wasi-http/src/body.rs
index 4456fe24c..03cc695bc 100644
--- a/crates/wasi-http/src/body.rs
+++ b/crates/wasi-http/src/body.rs
@@ -7,11 +7,7 @@ use http_body_util::BodyExt;
 use std::future::Future;
 use std::mem;
 use std::task::{Context, Poll};
-use std::{
-    pin::Pin,
-    sync::{Arc, Mutex},
-    time::Duration,
-};
+use std::{pin::Pin, sync::Arc, time::Duration};
 use tokio::sync::{mpsc, oneshot};
 use wasmtime_wasi::preview2::{
     self, poll_noop, AbortOnDropJoinHandle, HostInputStream, HostOutputStream, StreamError,
@@ -424,7 +420,7 @@ impl HostOutgoingBody {
             }
         }
 
-        let (body_sender, body_receiver) = mpsc::channel(1);
+        let (body_sender, body_receiver) = mpsc::channel(2);
         let (finish_sender, finish_receiver) = oneshot::channel();
         let body_impl = BodyImpl {
             body_receiver,
@@ -442,156 +438,20 @@ impl HostOutgoingBody {
     }
 }
 
-// copied in from preview2::write_stream
-
-#[derive(Debug)]
-struct WorkerState {
-    alive: bool,
-    items: std::collections::VecDeque<Bytes>,
-    write_budget: usize,
-    flush_pending: bool,
-    error: Option<anyhow::Error>,
-}
-
-impl WorkerState {
-    fn check_error(&mut self) -> Result<(), StreamError> {
-        if let Some(e) = self.error.take() {
-            return Err(StreamError::LastOperationFailed(e));
-        }
-        if !self.alive {
-            return Err(StreamError::Closed);
-        }
-        Ok(())
-    }
-}
-
-struct Worker {
-    state: Mutex<WorkerState>,
-    new_work: tokio::sync::Notify,
-    write_ready_changed: tokio::sync::Notify,
-}
-
-enum Job {
-    Flush,
-    Write(Bytes),
-}
-
-impl Worker {
-    fn new(write_budget: usize) -> Self {
-        Self {
-            state: Mutex::new(WorkerState {
-                alive: true,
-                items: std::collections::VecDeque::new(),
-                write_budget,
-                flush_pending: false,
-                error: None,
-            }),
-            new_work: tokio::sync::Notify::new(),
-            write_ready_changed: tokio::sync::Notify::new(),
-        }
-    }
-    async fn ready(&self) {
-        loop {
-            {
-                let state = self.state();
-                if state.error.is_some()
-                    || !state.alive
-                    || (!state.flush_pending && state.write_budget > 0)
-                {
-                    return;
-                }
-            }
-            self.write_ready_changed.notified().await;
-        }
-    }
-    fn check_write(&self) -> Result<usize, StreamError> {
-        let mut state = self.state();
-        if let Err(e) = state.check_error() {
-            return Err(e);
-        }
-
-        if state.flush_pending || state.write_budget == 0 {
-            return Ok(0);
-        }
-
-        Ok(state.write_budget)
-    }
-    fn state(&self) -> std::sync::MutexGuard<WorkerState> {
-        self.state.lock().unwrap()
-    }
-    fn pop(&self) -> Option<Job> {
-        let mut state = self.state();
-        if state.items.is_empty() {
-            if state.flush_pending {
-                return Some(Job::Flush);
-            }
-        } else if let Some(bytes) = state.items.pop_front() {
-            return Some(Job::Write(bytes));
-        }
-
-        None
-    }
-    fn report_error(&self, e: std::io::Error) {
-        {
-            let mut state = self.state();
-            state.alive = false;
-            state.error = Some(e.into());
-            state.flush_pending = false;
-        }
-        self.write_ready_changed.notify_one();
-    }
-
-    async fn work(&self, writer: mpsc::Sender<Bytes>) {
-        loop {
-            while let Some(job) = self.pop() {
-                match job {
-                    Job::Flush => {
-                        self.state().flush_pending = false;
-                    }
-
-                    Job::Write(bytes) => {
-                        tracing::debug!("worker writing: {bytes:?}");
-                        let len = bytes.len();
-                        match writer.send(bytes).await {
-                            Err(_) => {
-                                self.report_error(std::io::Error::new(
-                                    std::io::ErrorKind::BrokenPipe,
-                                    "Outgoing stream body reader has dropped",
-                                ));
-                                return;
-                            }
-                            Ok(_) => {
-                                self.state().write_budget += len;
-                            }
-                        }
-                    }
-                }
-
-                self.write_ready_changed.notify_one();
-            }
-
-            self.new_work.notified().await;
-        }
-    }
-}
-
 /// Provides a [`HostOutputStream`] impl from a [`tokio::sync::mpsc::Sender`].
 pub struct BodyWriteStream {
-    worker: Arc<Worker>,
-    _join_handle: preview2::AbortOnDropJoinHandle<()>,
+    writer: mpsc::Sender<Bytes>,
+    write_budget: usize,
 }
 
 impl BodyWriteStream {
     /// Create a [`BodyWriteStream`].
     pub fn new(write_budget: usize, writer: mpsc::Sender<Bytes>) -> Self {
-        let worker = Arc::new(Worker::new(write_budget));
-
-        let w = Arc::clone(&worker);
-        let join_handle = preview2::spawn(async move { w.work(writer).await });
-
+        // at least one capacity is required to send a message
+        assert!(writer.max_capacity() >= 1);
         BodyWriteStream {
-            worker,
-            _join_handle: join_handle,
+            writer,
+            write_budget,
         }
     }
 }
@@ -599,41 +459,57 @@ impl BodyWriteStream {
 #[async_trait::async_trait]
 impl HostOutputStream for BodyWriteStream {
     fn write(&mut self, bytes: Bytes) -> Result<(), StreamError> {
-        let mut state = self.worker.state();
-        state.check_error()?;
-        if state.flush_pending {
-            return Err(StreamError::Trap(anyhow!(
-                "write not permitted while flush pending"
-            )));
-        }
-        match state.write_budget.checked_sub(bytes.len()) {
-            Some(remaining_budget) => {
-                state.write_budget = remaining_budget;
-                state.items.push_back(bytes);
+        match self.writer.try_send(bytes) {
+            // If the message was sent then it's queued up now in hyper to get
+            // received.
+            Ok(()) => Ok(()),
+
+            // If this channel is full then that means `check_write` wasn't
+            // called. The call to `check_write` always guarantees that there's
+            // at least one capacity if a write is allowed.
+            Err(mpsc::error::TrySendError::Full(_)) => {
+                Err(StreamError::Trap(anyhow!("write exceeded budget")))
             }
-            None => return Err(StreamError::Trap(anyhow!("write exceeded budget"))),
+
+            // Hyper is gone so this stream is now closed.
+            Err(mpsc::error::TrySendError::Closed(_)) => Err(StreamError::Closed),
         }
-        drop(state);
-        self.worker.new_work.notify_one();
-        Ok(())
     }
-    fn flush(&mut self) -> Result<(), StreamError> {
-        let mut state = self.worker.state();
-        state.check_error()?;
 
-        state.flush_pending = true;
-        self.worker.new_work.notify_one();
-
-        Ok(())
+    fn flush(&mut self) -> Result<(), StreamError> {
+        // Flushing doesn't happen in this body stream since we're currently
+        // only tracking sending bytes over to hyper.
+        if self.writer.is_closed() {
+            Err(StreamError::Closed)
+        } else {
+            Ok(())
+        }
     }
 
     fn check_write(&mut self) -> Result<usize, StreamError> {
-        self.worker.check_write()
+        if self.writer.is_closed() {
+            Err(StreamError::Closed)
+        } else if self.writer.capacity() == 0 {
+            // If there is no more capacity in this sender channel then don't
+            // allow any more writes because the hyper task needs to catch up
+            // now.
+            //
+            // Note that this relies on this task being the only one sending
+            // data to ensure that no one else can steal a write into this
+            // channel.
+            Ok(0)
+        } else {
+            Ok(self.write_budget)
+        }
     }
 }
+
 #[async_trait::async_trait]
 impl Subscribe for BodyWriteStream {
     async fn ready(&mut self) {
-        self.worker.ready().await
+        // Attempt to perform a reservation for a send. If there's capacity in
+        // the channel or it's already closed then this will return immediately.
+        // If the channel is full this will block until capacity opens up.
+        let _ = self.writer.reserve().await;
     }
 }
diff --git a/crates/wasmtime/src/component/linker.rs b/crates/wasmtime/src/component/linker.rs
index 8efbbcbfe..5e490fa9a 100644
--- a/crates/wasmtime/src/component/linker.rs
+++ b/crates/wasmtime/src/component/linker.rs
@@ -378,7 +378,34 @@ impl<T> LinkerInstance<'_, T> {
         }
     }
 
-    // TODO: define func_new_async
+    /// Define a new host-provided async function using dynamic types.
+    ///
+    /// This is exactly like [`Self::func_new`] except it takes an async
+    /// host function.
+    #[cfg(feature = "async")]
+    #[cfg_attr(nightlydoc, doc(cfg(feature = "async")))]
+    pub fn func_new_async<F>(&mut self, component: &Component, name: &str, f: F) -> Result<()>
+    where
+        F: for<'a> Fn(
+                StoreContextMut<'a, T>,
+                &'a [Val],
+                &'a mut [Val],
+            ) -> Box<dyn Future<Output = Result<()>> + Send + 'a>
+            + Send
+            + Sync
+            + 'static,
+    {
+        assert!(
+            self.engine.config().async_support,
+            "cannot use `func_new_async` without enabling async support in the config"
+        );
+        let ff = move |mut store: StoreContextMut<'_, T>, params: &[Val], results: &mut [Val]| {
+            let async_cx = store.as_context_mut().0.async_cx().expect("async cx");
+            let mut future = Pin::from(f(store.as_context_mut(), params, results));
+            unsafe { async_cx.block_on(future.as_mut()) }?
+        };
+        self.func_new(component, name, ff)
+    }
 
     /// Defines a [`Module`] within this instance.
     ///


Test patch:
diff --git a/cranelift/filetests/filetests/isa/riscv64/arithmetic.clif b/cranelift/filetests/filetests/isa/riscv64/arithmetic.clif
index 070d7e178..1de5260db 100644
--- a/cranelift/filetests/filetests/isa/riscv64/arithmetic.clif
+++ b/cranelift/filetests/filetests/isa/riscv64/arithmetic.clif
@@ -512,14 +512,12 @@ block0(v0: i32):
 
 ; VCode:
 ; block0:
-;   li a3,-1
-;   subw a0,a0,a3
+;   addiw a0,a0,1
 ;   ret
 ;
 ; Disassembled:
 ; block0: ; offset 0x0
-;   addi a3, zero, -1
-;   subw a0, a0, a3
+;   addiw a0, a0, 1
 ;   ret
 
 function %f28(i64) -> i64 {
@@ -531,14 +529,12 @@ block0(v0: i64):
 
 ; VCode:
 ; block0:
-;   li a3,-1
-;   sub a0,a0,a3
+;   addi a0,a0,1
 ;   ret
 ;
 ; Disassembled:
 ; block0: ; offset 0x0
-;   addi a3, zero, -1
-;   sub a0, a0, a3
+;   addi a0, a0, 1
 ;   ret
 
 function %f29(i64) -> i64 {
diff --git a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_no_spectre_i32_access_0_offset.wat b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_no_spectre_i32_access_0_offset.wat
index 10e381b3c..d72cfed5a 100644
--- a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_no_spectre_i32_access_0_offset.wat
+++ b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_no_spectre_i32_access_0_offset.wat
@@ -42,15 +42,14 @@
 ;; function u0:0:
 ;; block0:
 ;;   slli a3,a0,32
-;;   srli a5,a3,32
-;;   ld a4,8(a2)
-;;   li a0,4
-;;   sub a4,a4,a0
-;;   bgtu a5,a4,taken(label3),not_taken(label1)
+;;   srli a4,a3,32
+;;   ld a3,8(a2)
+;;   addi a3,a3,-4
+;;   bgtu a4,a3,taken(label3),not_taken(label1)
 ;; block1:
-;;   ld a0,0(a2)
-;;   add a5,a0,a5
-;;   sw a1,0(a5)
+;;   ld a5,0(a2)
+;;   add a4,a5,a4
+;;   sw a1,0(a4)
 ;;   j label2
 ;; block2:
 ;;   ret
@@ -59,16 +58,15 @@
 ;;
 ;; function u0:1:
 ;; block0:
-;;   slli a3,a0,32
-;;   srli a5,a3,32
-;;   ld a4,8(a1)
-;;   li a0,4
-;;   sub a4,a4,a0
-;;   bgtu a5,a4,taken(label3),not_taken(label1)
+;;   slli a2,a0,32
+;;   srli a4,a2,32
+;;   ld a3,8(a1)
+;;   addi a3,a3,-4
+;;   bgtu a4,a3,taken(label3),not_taken(label1)
 ;; block1:
-;;   ld a0,0(a1)
-;;   add a5,a0,a5
-;;   lw a0,0(a5)
+;;   ld a5,0(a1)
+;;   add a4,a5,a4
+;;   lw a0,0(a4)
 ;;   j label2
 ;; block2:
 ;;   ret
diff --git a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_yes_spectre_i32_access_0_offset.wat b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_yes_spectre_i32_access_0_offset.wat
index fdac31e6d..0bb3e3c82 100644
--- a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_yes_spectre_i32_access_0_offset.wat
+++ b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i32_index_0_guard_yes_spectre_i32_access_0_offset.wat
@@ -41,36 +41,34 @@
 
 ;; function u0:0:
 ;; block0:
-;;   slli a3,a0,32
-;;   srli a3,a3,32
+;;   slli a0,a0,32
+;;   srli a3,a0,32
 ;;   ld a4,8(a2)
-;;   li a5,4
-;;   sub a4,a4,a5
+;;   addi a4,a4,-4
 ;;   sltu a4,a4,a3
 ;;   ld a2,0(a2)
 ;;   add a2,a2,a3
-;;   sub a0,zero,a4
-;;   not a3,a0
-;;   and a4,a2,a3
-;;   sw a1,0(a4)
+;;   sub a5,zero,a4
+;;   not a3,a5
+;;   and a3,a2,a3
+;;   sw a1,0(a3)
 ;;   j label1
 ;; block1:
 ;;   ret
 ;;
 ;; function u0:1:
 ;; block0:
-;;   slli a2,a0,32
-;;   srli a3,a2,32
-;;   ld a2,8(a1)
-;;   li a4,4
-;;   sub a2,a2,a4
-;;   sltu a2,a2,a3
-;;   ld a4,0(a1)
-;;   add a3,a4,a3
-;;   sub a0,zero,a2
-;;   not a2,a0
-;;   and a4,a3,a2
-;;   lw a0,0(a4)
+;;   slli a0,a0,32
+;;   srli a2,a0,32
+;;   ld a3,8(a1)
+;;   addi a3,a3,-4
+;;   sltu a3,a3,a2
+;;   ld a1,0(a1)
+;;   add a1,a1,a2
+;;   sub a5,zero,a3
+;;   not a2,a5
+;;   and a3,a1,a2
+;;   lw a0,0(a3)
 ;;   j label1
 ;; block1:
 ;;   ret
diff --git a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_no_spectre_i32_access_0_offset.wat b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_no_spectre_i32_access_0_offset.wat
index 0b3cd6932..f21969e0c 100644
--- a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_no_spectre_i32_access_0_offset.wat
+++ b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_no_spectre_i32_access_0_offset.wat
@@ -42,13 +42,12 @@
 ;; function u0:0:
 ;; block0:
 ;;   ld a3,8(a2)
-;;   li a4,4
-;;   sub a3,a3,a4
+;;   addi a3,a3,-4
 ;;   bgtu a0,a3,taken(label3),not_taken(label1)
 ;; block1:
-;;   ld a3,0(a2)
-;;   add a3,a3,a0
-;;   sw a1,0(a3)
+;;   ld a2,0(a2)
+;;   add a2,a2,a0
+;;   sw a1,0(a2)
 ;;   j label2
 ;; block2:
 ;;   ret
@@ -58,13 +57,12 @@
 ;; function u0:1:
 ;; block0:
 ;;   ld a2,8(a1)
-;;   li a3,4
-;;   sub a2,a2,a3
+;;   addi a2,a2,-4
 ;;   bgtu a0,a2,taken(label3),not_taken(label1)
 ;; block1:
-;;   ld a3,0(a1)
-;;   add a3,a3,a0
-;;   lw a0,0(a3)
+;;   ld a2,0(a1)
+;;   add a2,a2,a0
+;;   lw a0,0(a2)
 ;;   j label2
 ;; block2:
 ;;   ret
diff --git a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_yes_spectre_i32_access_0_offset.wat b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_yes_spectre_i32_access_0_offset.wat
index f98c4f604..76bca6405 100644
--- a/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_yes_spectre_i32_access_0_offset.wat
+++ b/cranelift/filetests/filetests/isa/riscv64/wasm/load_store_dynamic_kind_i64_index_0_guard_yes_spectre_i32_access_0_offset.wat
@@ -41,15 +41,14 @@
 
 ;; function u0:0:
 ;; block0:
-;;   ld a3,8(a2)
-;;   li a4,4
-;;   sub a3,a3,a4
-;;   sltu a3,a3,a0
+;;   ld a5,8(a2)
+;;   addi a5,a5,-4
+;;   sltu a5,a5,a0
 ;;   ld a2,0(a2)
 ;;   add a0,a2,a0
-;;   sub a4,zero,a3
-;;   not a2,a4
-;;   and a2,a0,a2
+;;   sub a3,zero,a5
+;;   not a5,a3
+;;   and a2,a0,a5
 ;;   sw a1,0(a2)
 ;;   j label1
 ;; block1:
@@ -57,16 +56,15 @@
 ;;
 ;; function u0:1:
 ;; block0:
-;;   ld a2,8(a1)
-;;   li a3,4
-;;   sub a2,a2,a3
-;;   sltu a2,a2,a0
+;;   ld a5,8(a1)
+;;   addi a5,a5,-4
+;;   sltu a5,a5,a0
 ;;   ld a1,0(a1)
 ;;   add a0,a1,a0
-;;   sub a4,zero,a2
-;;   not a1,a4
-;;   and a2,a0,a1
-;;   lw a0,0(a2)
+;;   sub a3,zero,a5
+;;   not a5,a3
+;;   and a1,a0,a5
+;;   lw a0,0(a1)
 ;;   j label1
 ;; block1:
 ;;   ret
diff --git a/cranelift/filetests/filetests/isa/riscv64/zca.clif b/cranelift/filetests/filetests/isa/riscv64/zca.clif
index c0cf006d6..44feb0957 100644
--- a/cranelift/filetests/filetests/isa/riscv64/zca.clif
+++ b/cranelift/filetests/filetests/isa/riscv64/zca.clif
@@ -906,3 +906,19 @@ block0(v0: i64):
 ;   c.ld a0, 0x40(a0)
 ;   c.jr ra
 
+function %band_reverse(i64, i64) -> i64 {
+block0(v0: i64, v1: i64):
+    v2 = band v1, v0
+    return v2
+}
+
+; VCode:
+; block0:
+;   and a0,a1,a0
+;   ret
+;
+; Disassembled:
+; block0: ; offset 0x0
+;   c.and a0, a1
+;   c.jr ra
+