feat: update hyper deps to 1.4

Author: ruslanti

Cargo.toml

@@ -3,7 +3,7 @@ members = ["crates/*"]
 resolver = "2"
 
 [workspace.package]
-version = "0.4.2"
+version = "0.5.0"
 edition = "2021"
 publish = false
 authors = ["FastEdge Development Team"]
@@ -15,10 +15,10 @@ tracing = "0.1"
 hyper = { version = "1", features = ["full"] }
 http = "1.1.0"
 async-trait = "0.1"
-wasmtime = { version = "22.0" }
-wasmtime-wasi = { version = "22.0" }
-wasi-common = { version = "22.0" }
-wasmtime-wasi-nn = { version = "22.0" }
+wasmtime = { version = "20.0.2" }
+wasmtime-wasi = { version = "20.0.2" }
+wasi-common = { version = "20.0.2" }
+wasmtime-wasi-nn = { version = "20.0.2" }
 clap = { version = "4", features = ["derive"] }
 moka = { version = "0.12", features = ["sync"] }
 smol_str = { version = "0.2.1", features = ["serde"] }

crates/candle-wasi-nn/Cargo.toml

@@ -5,6 +5,10 @@ edition.workspace = true
 publish.workspace = true
 authors.workspace = true
 
+[features]
+metal = []
+cuda = []
+
 [dependencies]
 wasmtime-wasi-nn = { workspace = true }
 tracing = { workspace = true }

crates/http-backend/src/lib.rs

@@ -7,13 +7,13 @@ use std::time::Duration;
 use anyhow::{anyhow, Error, Result};
 use async_trait::async_trait;
 use http::{uri::Scheme, HeaderMap, HeaderValue, Uri};
-use http_body_util::Full;
+use http_body_util::{BodyExt, Full};
 use hyper::body::Bytes;
+use hyper::rt::ReadBufCursor;
 use hyper_util::client::legacy::connect::{Connect, HttpConnector};
 use hyper_util::client::legacy::Client;
 use hyper_util::rt::TokioExecutor;
 use pin_project::pin_project;
-use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::TcpStream;
 use tower_service::Service;
 use tracing::{debug, trace, warn};
@@ -264,9 +264,9 @@ where
         let request = self.make_request(req)?;
         let res = self.client.request(request).await?;
 
-        let _status = res.status().as_u16();
-        let (parts, _body) = res.into_parts();
-        let _headers = if !parts.headers.is_empty() {
+        let status = res.status().as_u16();
+        let (parts, body) = res.into_parts();
+        let headers = if !parts.headers.is_empty() {
             Some(
                 parts
                     .headers
@@ -284,7 +284,7 @@ where
             None
         };
 
-        /*let body_bytes = body.into();
+        let body_bytes = body.collect().await?.to_bytes();
         let body = Some(body_bytes.to_vec());
 
         trace!(?status, ?headers, len = body_bytes.len(), "reply");
@@ -293,8 +293,7 @@ where
             status,
             headers,
             body,
-        }))*/
-        unimplemented!("send request")
+        }))
     }
 }
 
@@ -399,41 +398,34 @@ impl Service<Uri> for FastEdgeConnector {
     }
 }
 
-impl AsyncRead for Connection {
-    fn poll_read(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut ReadBuf<'_>,
-    ) -> Poll<std::io::Result<()>> {
-        let this = self.project();
-        this.inner.poll_read(cx, buf)
+impl hyper::rt::Read for Connection {
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, mut buf: ReadBufCursor<'_>) -> Poll<std::result::Result<(), std::io::Error>> {
+        let n = unsafe {
+            let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
+            match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
+                Poll::Ready(Ok(())) => tbuf.filled().len(),
+                other => return other,
+            }
+        };
+
+        unsafe {
+            buf.advance(n);
+        }
+        Poll::Ready(Ok(()))
     }
 }
 
-impl AsyncWrite for Connection {
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<std::result::Result<usize, std::io::Error>> {
-        let this = self.project();
-        this.inner.poll_write(cx, buf)
+impl hyper::rt::Write for Connection {
+    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::result::Result<usize, std::io::Error>> {
+        tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
     }
 
-    fn poll_flush(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<std::result::Result<(), std::io::Error>> {
-        let this = self.project();
-        this.inner.poll_flush(cx)
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::result::Result<(), std::io::Error>> {
+        tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
     }
 
-    fn poll_shutdown(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<std::result::Result<(), std::io::Error>> {
-        let this = self.project();
-        this.inner.poll_shutdown(cx)
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::result::Result<(), std::io::Error>> {
+        tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
     }
 }
 

crates/http-service/Cargo.toml

@@ -9,7 +9,7 @@ authors.workspace = true
 default = []
 metrics = ["runtime/metrics"]
 stats = ["runtime/stats"]
-tls = ["tokio-rustls", "rustls-pemfile", "hyper-rustls", "rustls"]
+#tls = ["tokio-rustls", "rustls-pemfile", "hyper-rustls", "rustls"]
 
 [dependencies]
 anyhow = { workspace = true }
@@ -23,10 +23,6 @@ wasmtime-wasi-nn = { workspace = true }
 wasi-common = { workspace = true }
 tracing = { workspace = true }
 smol_str = { workspace = true }
-tokio-rustls = { version = "0.24.1", optional = true}
-rustls-pemfile = { version = "1.0.2" , optional = true}
-hyper-rustls = { version = "0.24.1", optional = true }
-rustls = { version = "0.21.6", optional = true }
 reactor = { path = "../reactor" }
 runtime = { path = "../runtime" }
 http-backend = { path = "../http-backend" }
@@ -40,9 +36,10 @@ clickhouse = { version = "0.12.0", optional = true }
 chrono = "0.4"
 async-trait = "0.1"
 wasmtime-wasi-http = "20.0.2"
-hyper-util = "0.1.3"
-http-body-util = "0.1.1"
+hyper-util = "0.1"
+http-body-util = "0.1"
 shellflip = {workspace = true}
+bytes = "1.6"
 
 [dev-dependencies]
 claims = "0.7"

crates/http-service/src/executor/mod.rs

@@ -6,9 +6,10 @@ use std::time::{Duration, Instant};
 use anyhow::{anyhow, bail, Context, Error, Result};
 use async_trait::async_trait;
 use bytesize::ByteSize;
-use http::{HeaderMap, HeaderValue, Method};
+use http::{HeaderMap, HeaderValue, Method, Response};
 use http_backend::Backend;
 use http_body_util::{BodyExt, Full};
+use http_body_util::combinators::BoxBody;
 use hyper::body::{Bytes, Incoming};
 use smol_str::SmolStr;
 use wasmtime_wasi::StdoutStream;
@@ -30,7 +31,7 @@ pub trait HttpExecutor {
     async fn execute(
         &self,
         req: hyper::Request<Incoming>,
-    ) -> Result<(hyper::Response<Full<Bytes>>, Duration, ByteSize)>;
+    ) -> Result<(Response<BoxBody<Bytes, hyper::Error>>, Duration, ByteSize)>;
 }
 
 pub trait ExecutorFactory<C> {
@@ -46,7 +47,7 @@ pub trait ExecutorFactory<C> {
 /// Execute context used by ['HttpService']
 #[derive(Clone)]
 pub struct HttpExecutorImpl<C> {
-    instance_pre: InstancePre<HttpState>,
+    instance_pre: InstancePre<HttpState<C>>,
     store_builder: StoreBuilder,
     backend: Backend<C>,
 }
@@ -59,7 +60,7 @@ where
     async fn execute(
         &self,
         req: hyper::Request<Incoming>,
-    ) -> Result<(hyper::Response<Full<Bytes>>, Duration, ByteSize)> {
+    ) -> Result<(Response<BoxBody<Bytes, hyper::Error>>, Duration, ByteSize)> {
         let start_ = Instant::now();
         let response = self.execute_impl(req).await;
         let elapsed = Instant::now().duration_since(start_);
@@ -72,7 +73,7 @@ where
     C: Clone + Send + Sync + 'static,
 {
     pub fn new(
-        instance_pre: InstancePre<HttpState>,
+        instance_pre: InstancePre<HttpState<C>>,
         store_builder: StoreBuilder,
         backend: Backend<C>,
     ) -> Self {
@@ -86,7 +87,7 @@ where
     async fn execute_impl(
         &self,
         req: hyper::Request<Incoming>,
-    ) -> Result<(hyper::Response<Full<Bytes>>, ByteSize)> {
+    ) -> Result<(Response<BoxBody<Bytes, hyper::Error>>, ByteSize)> {
         let (parts, body) = req.into_parts();
         let method = to_fastedge_http_method(&parts.method)?;
 
@@ -125,7 +126,7 @@ where
             .propagate_headers(&parts.headers)
             .context("propagate headers")?;
 
-        let state = HttpState { wasi_nn };
+        let state = HttpState { wasi_nn, http_backend };
 
         let mut store = store_builder.build(state)?;
 
@@ -158,7 +159,7 @@ where
         };
         let used = ByteSize::b(store.memory_used() as u64);
 
-        let body = resp.body.map(|b| Full::from(b)).unwrap_or_default();
+        let body = resp.body.map(|b| Full::from(b).map_err(|never| match never {}).boxed()).unwrap_or_default();
         builder.body(body).map(|r| (r, used)).map_err(Error::msg)
     }
 

crates/http-service/src/executor/wasi_http.rs

@@ -1,13 +1,16 @@
 use std::time::{Duration, Instant};
 
-use anyhow::bail;
+use anyhow::{bail, Context};
 use async_trait::async_trait;
 use bytesize::ByteSize;
+use http::Response;
 use http_body_util::{BodyExt, Full};
+use http_body_util::combinators::BoxBody;
 use hyper::body::{Bytes, Incoming};
 use tracing::{error, info};
 use wasmtime_wasi_http::WasiHttpView;
 
+use http_backend::Backend;
 use runtime::InstancePre;
 use runtime::store::StoreBuilder;
 
@@ -16,36 +19,44 @@ use crate::HttpState;
 
 /// Execute context used by ['HttpService']
 #[derive(Clone)]
-pub struct WasiHttpExecutorImpl {
-    instance_pre: InstancePre<HttpState>,
+pub struct WasiHttpExecutorImpl<C> {
+    instance_pre: InstancePre<HttpState<C>>,
     store_builder: StoreBuilder,
+    backend: Backend<C>,
 }
 
 #[async_trait]
-impl HttpExecutor for WasiHttpExecutorImpl {
+impl<C> HttpExecutor for WasiHttpExecutorImpl<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
     async fn execute(
         &self,
         req: hyper::Request<Incoming>,
-    ) -> anyhow::Result<(hyper::Response<Full<Bytes>>, Duration, ByteSize)> {
+    ) -> anyhow::Result<(Response<BoxBody<Bytes, hyper::Error>>, Duration, ByteSize)> {
         let start_ = Instant::now();
         let response = self.execute_impl(req).await;
         let elapsed = Instant::now().duration_since(start_);
         response.map(|(r, used)| (r, elapsed, used))
     }
 }
 
-impl WasiHttpExecutorImpl {
-    pub fn new(instance_pre: InstancePre<HttpState>, store_builder: StoreBuilder) -> Self {
+impl<C> WasiHttpExecutorImpl<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    pub fn new(instance_pre: InstancePre<HttpState<C>>, store_builder: StoreBuilder, backend: Backend<C>) -> Self {
         Self {
             instance_pre,
             store_builder,
+            backend,
         }
     }
 
     async fn execute_impl(
         &self,
         req: hyper::Request<Incoming>,
-    ) -> anyhow::Result<(hyper::Response<Full<Bytes>>, ByteSize)> {
+    ) -> anyhow::Result<(Response<BoxBody<Bytes, hyper::Error>>, ByteSize)> {
         let (sender, receiver) = tokio::sync::oneshot::channel();
         let (parts, body) = req.into_parts();
 
@@ -58,8 +69,13 @@ impl WasiHttpExecutorImpl {
             .store_builder
             .make_wasi_nn_ctx()
             .expect("make_wasi_nn_ctx");
+        let mut http_backend = self.backend.to_owned();
 
-        let state = HttpState { wasi_nn };
+        http_backend
+            .propagate_headers(&parts.headers)
+            .context("propagate headers")?;
+
+        let state = HttpState { wasi_nn, http_backend };
 
         let mut store = store_builder.build(state).expect("store build");
         let instance_pre = self.instance_pre.clone();
@@ -106,9 +122,9 @@ impl WasiHttpExecutorImpl {
             Ok(Ok(resp)) => {
                 let (parts, body) = resp.into_parts();
                 let body = body.collect().await.expect("incoming body").to_bytes();
-                let body = Full::new(body);
+                let body = Full::new(body).map_err(|never| match never {}).boxed();
                 let used = task.await.expect("task await").expect("byte size");
-                Ok((hyper::Response::from_parts(parts, body), used))
+                Ok((Response::from_parts(parts, body), used))
             }
             Ok(Err(e)) => Err(e.into()),
             Err(_) => {