Repository: bytecodealliance/wasmtime
Pull request #9208 (2024-09-06 16:55:49)

[wasi-http] Return `none` from `future-trailers.get` when trailers absent
Previously, we would return `error-code::connection-terminated` in this case, whereas the `wasi-http` spec says we should simply return `none` (i.e. it's not an error if there are no trailers present).

This also adds code to `api_proxy_streaming.rs` to check for trailers after reading the body content.

<!--
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
-->



Patch:
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index e494c98e3..46d2ea54d 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -686,6 +686,11 @@ jobs:
           echo "::warning::This CI run will not test MPK; it has been detected as not available on this machine (\`cargo run --example mpk-available\`)."
         fi
 
+    # Install VTune, see `cli_tests::profile_with_vtune`.
+    - name: Install VTune
+      if: matrix.filter == 'linux-x64' && contains(matrix.bucket, 'wasmtime-cli')
+      uses: abrown/install-vtune-action@v1
+
     # Build and test all features
     - run: ./ci/run-tests.sh --locked ${{ matrix.bucket }}
       env:
diff --git a/cranelift/codegen/src/isa/aarch64/lower.isle b/cranelift/codegen/src/isa/aarch64/lower.isle
index 415d40854..7d18b602e 100644
--- a/cranelift/codegen/src/isa/aarch64/lower.isle
+++ b/cranelift/codegen/src/isa/aarch64/lower.isle
@@ -820,6 +820,26 @@
            (dst_lo Reg (madd $I64 x_lo y_lo (zero_reg))))
         (value_regs dst_lo dst_hi)))
 
+;; Special cases where the upper bits are sign-or-zero extended of the lower bits
+;; so the calculation here is much simpler with just a `umulh` or `smulh`
+;; instead of the additions above as well.
+(rule (lower (has_type $I128 (imul (uextend x) (uextend y))))
+      (let (
+          (x Reg (put_in_reg_zext64 x))
+          (y Reg (put_in_reg_zext64 y))
+        )
+        (value_regs
+          (madd $I64 x y (zero_reg))
+          (umulh $I64 x y))))
+(rule (lower (has_type $I128 (imul (sextend x) (sextend y))))
+      (let (
+          (x Reg (put_in_reg_sext64 x))
+          (y Reg (put_in_reg_sext64 y))
+        )
+        (value_regs
+          (madd $I64 x y (zero_reg))
+          (smulh $I64 x y))))
+
 ;; Case for i8x16, i16x8, and i32x4.
 (rule -2 (lower (has_type (ty_vec128 ty @ (not_i64x2)) (imul x y)))
       (mul x y (vector_size ty)))
diff --git a/crates/wasi-http/src/body.rs b/crates/wasi-http/src/body.rs
index 5f16f4239..5913a4922 100644
--- a/crates/wasi-http/src/body.rs
+++ b/crates/wasi-http/src/body.rs
@@ -347,12 +347,9 @@ impl Subscribe for HostFutureTrailers {
                 // were reached. It's up to us now to complete the body.
                 Ok(StreamEnd::Remaining(b)) => body.body = IncomingBodyState::Start(b),
 
-                // Technically this shouldn't be possible as the sender
-                // shouldn't get destroyed without receiving a message. Handle
-                // this just in case though.
+                // This means there were no trailers present.
                 Err(_) => {
-                    debug_assert!(false, "should be unreachable");
-                    *self = HostFutureTrailers::Done(Err(types::ErrorCode::ConnectionTerminated));
+                    *self = HostFutureTrailers::Done(Ok(None));
                 }
             }
         }


Test patch:
diff --git a/cranelift/filetests/filetests/isa/aarch64/i128.clif b/cranelift/filetests/filetests/isa/aarch64/i128.clif
new file mode 100644
index 000000000..bee69e094
--- /dev/null
+++ b/cranelift/filetests/filetests/isa/aarch64/i128.clif
@@ -0,0 +1,48 @@
+test compile precise-output
+set enable_llvm_abi_extensions=true
+target aarch64
+
+function %mul_uextend_i64(i64, i64) -> i128 {
+block0(v0: i64, v1: i64):
+    v2 = uextend.i128 v0
+    v3 = uextend.i128 v1
+    v4 = imul v2, v3
+    return v4
+}
+
+; VCode:
+; block0:
+;   madd x3, x0, x1, xzr
+;   umulh x1, x0, x1
+;   mov x0, x3
+;   ret
+;
+; Disassembled:
+; block0: ; offset 0x0
+;   mul x3, x0, x1
+;   umulh x1, x0, x1
+;   mov x0, x3
+;   ret
+
+function %mul_sextend_i64(i64, i64) -> i128 {
+block0(v0: i64, v1: i64):
+    v2 = sextend.i128 v0
+    v3 = sextend.i128 v1
+    v4 = imul v2, v3
+    return v4
+}
+
+; VCode:
+; block0:
+;   madd x3, x0, x1, xzr
+;   smulh x1, x0, x1
+;   mov x0, x3
+;   ret
+;
+; Disassembled:
+; block0: ; offset 0x0
+;   mul x3, x0, x1
+;   smulh x1, x0, x1
+;   mov x0, x3
+;   ret
+
diff --git a/crates/test-programs/src/bin/api_proxy_streaming.rs b/crates/test-programs/src/bin/api_proxy_streaming.rs
index 328e8a12d..39b178272 100644
--- a/crates/test-programs/src/bin/api_proxy_streaming.rs
+++ b/crates/test-programs/src/bin/api_proxy_streaming.rs
@@ -318,8 +318,8 @@ mod executor {
         http::{
             outgoing_handler,
             types::{
-                self, IncomingBody, IncomingResponse, InputStream, OutgoingBody, OutgoingRequest,
-                OutputStream,
+                self, FutureTrailers, IncomingBody, IncomingResponse, InputStream, OutgoingBody,
+                OutgoingRequest, OutputStream,
             },
         },
         io::{self, streams::StreamError},
@@ -465,42 +465,102 @@ mod executor {
     }
 
     pub fn incoming_body(body: IncomingBody) -> impl Stream<Item = Result<Vec<u8>>> {
-        struct Incoming(Option<(InputStream, IncomingBody)>);
+        enum Inner {
+            Stream {
+                stream: InputStream,
+                body: IncomingBody,
+            },
+            Trailers(FutureTrailers),
+            Closed,
+        }
+
+        struct Incoming(Inner);
 
         impl Drop for Incoming {
             fn drop(&mut self) {
-                if let Some((stream, body)) = self.0.take() {
-                    drop(stream);
-                    IncomingBody::finish(body);
+                match mem::replace(&mut self.0, Inner::Closed) {
+                    Inner::Stream { stream, body } => {
+                        drop(stream);
+                        IncomingBody::finish(body);
+                    }
+                    Inner::Trailers(_) | Inner::Closed => {}
                 }
             }
         }
 
         stream::poll_fn({
             let stream = body.stream().expect("response body should be readable");
-            let pair = Incoming(Some((stream, body)));
+            let mut incoming = Incoming(Inner::Stream { stream, body });
 
             move |context| {
-                if let Some((stream, _)) = &pair.0 {
-                    match stream.read(READ_SIZE) {
-                        Ok(buffer) => {
-                            if buffer.is_empty() {
-                                WAKERS
-                                    .lock()
-                                    .unwrap()
-                                    .push((stream.subscribe(), context.waker().clone()));
-                                Poll::Pending
-                            } else {
-                                Poll::Ready(Some(Ok(buffer)))
+                loop {
+                    match &incoming.0 {
+                        Inner::Stream { stream, .. } => match stream.read(READ_SIZE) {
+                            Ok(buffer) => {
+                                return if buffer.is_empty() {
+                                    WAKERS
+                                        .lock()
+                                        .unwrap()
+                                        .push((stream.subscribe(), context.waker().clone()));
+                                    Poll::Pending
+                                } else {
+                                    Poll::Ready(Some(Ok(buffer)))
+                                };
+                            }
+                            Err(StreamError::Closed) => {
+                                let Inner::Stream { stream, body } =
+                                    mem::replace(&mut incoming.0, Inner::Closed)
+                                else {
+                                    unreachable!();
+                                };
+                                drop(stream);
+                                incoming.0 = Inner::Trailers(IncomingBody::finish(body));
+                            }
+                            Err(StreamError::LastOperationFailed(error)) => {
+                                return Poll::Ready(Some(Err(anyhow!(
+                                    "{}",
+                                    error.to_debug_string()
+                                ))));
+                            }
+                        },
+
+                        Inner::Trailers(trailers) => {
+                            match trailers.get() {
+                                Some(Ok(trailers)) => {
+                                    incoming.0 = Inner::Closed;
+                                    match trailers {
+                                        Ok(Some(_)) => {
+                                            // Currently, we just ignore any trailers.  TODO: Add a test that
+                                            // expects trailers and verify they match the expected contents.
+                                        }
+                                        Ok(None) => {
+                                            // No trailers; nothing else to do.
+                                        }
+                                        Err(error) => {
+                                            // Error reading the trailers: pass it on to the application.
+                                            return Poll::Ready(Some(Err(anyhow!("{error:?}"))));
+                                        }
+                                    }
+                                }
+                                Some(Err(_)) => {
+                                    // Should only happen if we try to retrieve the trailers twice, i.e. a bug in
+                                    // this code.
+                                    unreachable!();
+                                }
+                                None => {
+                                    WAKERS
+                                        .lock()
+                                        .unwrap()
+                                        .push((trailers.subscribe(), context.waker().clone()));
+                                    return Poll::Pending;
+                                }
                             }
                         }
-                        Err(StreamError::Closed) => Poll::Ready(None),
-                        Err(StreamError::LastOperationFailed(error)) => {
-                            Poll::Ready(Some(Err(anyhow!("{}", error.to_debug_string()))))
+
+                        Inner::Closed => {
+                            return Poll::Ready(None);
                         }
                     }
-                } else {
-                    Poll::Ready(None)
                 }
             }
         })
diff --git a/tests/all/cli_tests.rs b/tests/all/cli_tests.rs
index dfb31842a..833efe56e 100644
--- a/tests/all/cli_tests.rs
+++ b/tests/all/cli_tests.rs
@@ -3,7 +3,7 @@
 use anyhow::{bail, Result};
 use std::fs::File;
 use std::io::Write;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::process::{Command, Output, Stdio};
 use tempfile::{NamedTempFile, TempDir};
 
@@ -21,13 +21,10 @@ pub fn run_wasmtime_for_output(args: &[&str], stdin: Option<&Path>) -> Result<Ou
 /// Get the Wasmtime CLI as a [Command].
 pub fn get_wasmtime_command() -> Result<Command> {
     // Figure out the Wasmtime binary from the current executable.
+    let bin = get_wasmtime_path()?;
     let runner = std::env::vars()
         .filter(|(k, _v)| k.starts_with("CARGO_TARGET") && k.ends_with("RUNNER"))
         .next();
-    let mut me = std::env::current_exe()?;
-    me.pop(); // chop off the file name
-    me.pop(); // chop off `deps`
-    me.push("wasmtime");
 
     // If we're running tests with a "runner" then we might be doing something
     // like cross-emulation, so spin up the emulator rather than the tests
@@ -38,10 +35,10 @@ pub fn get_wasmtime_command() -> Result<Command> {
         for arg in parts {
             cmd.arg(arg);
         }
-        cmd.arg(&me);
+        cmd.arg(&bin);
         cmd
     } else {
-        Command::new(&me)
+        Command::new(&bin)
     };
 
     // Ignore this if it's specified in the environment to allow tests to run in
@@ -51,6 +48,14 @@ pub fn get_wasmtime_command() -> Result<Command> {
     Ok(cmd)
 }
 
+fn get_wasmtime_path() -> Result<PathBuf> {
+    let mut path = std::env::current_exe()?;
+    path.pop(); // chop off the file name
+    path.pop(); // chop off `deps`
+    path.push("wasmtime");
+    Ok(path)
+}
+
 // Run the wasmtime CLI with the provided args and, if it succeeds, return
 // the standard output in a `String`.
 fn run_wasmtime(args: &[&str]) -> Result<String> {
@@ -1947,3 +1952,43 @@ fn settings_command() -> Result<()> {
     assert!(output.contains("Cranelift settings for target"));
     Ok(())
 }
+
+#[cfg(target_arch = "x86_64")]
+#[test]
+fn profile_with_vtune() -> Result<()> {
+    if !is_vtune_available() {
+        println!("> `vtune` is not available on the system path; skipping test");
+        return Ok(());
+    }
+
+    let mut bin = Command::new("vtune");
+    bin.args(&[
+        // Configure VTune...
+        "-verbose",
+        "-collect",
+        "hotspots",
+        "-user-data-dir",
+        &std::env::temp_dir().to_string_lossy(),
+        // ...then run Wasmtime with profiling enabled:
+        &get_wasmtime_path()?.to_string_lossy(),
+        "--profile=vtune",
+        "tests/all/cli_tests/simple.wat",
+    ]);
+
+    println!("> executing: {bin:?}");
+    let output = bin.output()?;
+
+    assert!(output.status.success());
+    let stdout = String::from_utf8_lossy(&output.stdout);
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    println!("> stdout:\n{stdout}");
+    assert!(stdout.contains("CPU Time"));
+    println!("> stderr:\n{stderr}");
+    assert!(!stderr.contains("Error"));
+    Ok(())
+}
+
+#[cfg(target_arch = "x86_64")]
+fn is_vtune_available() -> bool {
+    Command::new("vtune").arg("-version").output().is_ok()
+}