Merge pull request #15 from G-Core/feat/wasi-http

WASI HTTP Support

Author: ruslanti

.github/workflows/ci.yaml

@@ -31,8 +31,11 @@ jobs:
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
 
-      - name: Release Build
-        run: cargo build --release --all-features
+      - name: Build
+        run: cargo build --all-features
+
+      - name: Unit Tests
+        run: cargo test --all --exclude http-backend
 
       - name: Run Clippy
         run: cargo clippy --all-targets --all-features

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"]
@@ -12,13 +12,13 @@ authors = ["FastEdge Development Team"]
 tokio = { version = "1", features = ["full"] }
 tokio-util = { version = "0.7", features = ["codec"] }
 tracing = "0.1"
-hyper = { version = "0.14", features = ["full"] }
-http = "0.2.9"
+hyper = { version = "1", features = ["full"] }
+http = "1.1.0"
 async-trait = "0.1"
-wasmtime = { version = "20.0" }
-wasmtime-wasi = { version = "20.0" }
-wasi-common = { version = "20.0" }
-wasmtime-wasi-nn = { version = "20.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"] }
@@ -51,10 +51,15 @@ tokio-util = { workspace = true }
 wasmtime = { workspace = true }
 wasmtime-wasi = { workspace = true }
 smol_str = { workspace = true }
-clap = { version = "4.5.0", features = ["derive"] }
-pretty_env_logger = "0.5.0"
+async-trait = {workspace = true}
+clap = { version = "4.5", features = ["derive"] }
+pretty_env_logger = "0.5"
 runtime = { path = "crates/runtime", default-features = false }
 http-service = { path = "crates/http-service" }
 http-backend = { path = "crates/http-backend" }
-hyper-tls = "0.5.0"
+hyper-tls = "0.6"
+hyper-util = { version = "0.1", features = ["client", "client-legacy", "http1", "tokio"] }
+http-body-util = "0.1"
 shellflip = {workspace = true}
+bytesize = "1.3.0"
+

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/Cargo.toml

@@ -14,9 +14,13 @@ anyhow = {workspace = true}
 tracing = {workspace = true}
 hyper = { workspace = true }
 tokio = { workspace = true }
+hyper-util = { version = "0.1.3", features = ["client", "client-legacy", "http1", "tokio"] }
+http-body-util = "0.1.1"
 pin-project = "1.1.3"
 log = "0.4.20"
 url = "2.5.0"
+tower-service = "0.3.2"
+smol_str = {workspace = true}
 
 [dev-dependencies]
 claims = "0.7"

crates/http-backend/src/lib.rs

@@ -6,12 +6,17 @@ use std::time::Duration;
 
 use anyhow::{anyhow, Error, Result};
 use async_trait::async_trait;
-use http::{uri::Scheme, HeaderMap, HeaderValue, Uri};
-use hyper::client::connect::Connect;
-use hyper::{client::HttpConnector, service::Service, Client};
+use http::{header, uri::Scheme, HeaderMap, HeaderName, Uri};
+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 smol_str::{SmolStr, ToSmolStr};
 use tokio::net::TcpStream;
+use tower_service::Service;
 use tracing::{debug, trace, warn};
 
 use reactor::gcore::fastedge::http::Headers;
@@ -20,9 +25,7 @@ use reactor::gcore::fastedge::{
     http_client::Host,
 };
 
-const HOST_HEADER_NAME: &str = "host";
-
-type HeaderList = Vec<(String, String)>;
+type HeaderNameList = Vec<SmolStr>;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum BackendStrategy {
@@ -47,17 +50,17 @@ pub struct FastEdgeConnector {
 
 #[derive(Clone, Debug)]
 pub struct Backend<C> {
-    client: Client<C>,
+    client: Client<C, Full<Bytes>>,
     uri: Uri,
-    propagate_headers: HeaderList,
-    propagate_header_names: Vec<String>,
+    propagate_headers: HeaderMap,
+    propagate_header_names: HeaderNameList,
     max_sub_requests: usize,
     strategy: BackendStrategy,
 }
 
 pub struct Builder {
     uri: Uri,
-    propagate_header_names: Vec<String>,
+    propagate_header_names: HeaderNameList,
     max_sub_requests: usize,
     strategy: BackendStrategy,
 }
@@ -67,7 +70,7 @@ impl Builder {
         self.uri = uri;
         self
     }
-    pub fn propagate_headers_names(&mut self, propagate: Vec<String>) -> &mut Self {
+    pub fn propagate_headers_names(&mut self, propagate: HeaderNameList) -> &mut Self {
         self.propagate_header_names = propagate;
         self
     }
@@ -80,15 +83,15 @@ impl Builder {
     where
         C: Connect + Clone,
     {
-        let client = hyper::Client::builder()
+        let client = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
             .set_host(false)
             .pool_idle_timeout(Duration::from_secs(30))
             .build(connector);
 
         Backend {
             client,
             uri: self.uri.to_owned(),
-            propagate_headers: vec![],
+            propagate_headers: HeaderMap::new(),
             propagate_header_names: self.propagate_header_names.to_owned(),
             max_sub_requests: self.max_sub_requests,
             strategy: self.strategy,
@@ -106,43 +109,57 @@ impl<C> Backend<C> {
         }
     }
 
-    pub fn uri(&self) -> &Uri {
-        &self.uri
+    pub fn uri(&self) -> Uri {
+        self.uri.to_owned()
+    }
+
+    pub fn propagate_header_names(&self) -> Vec<SmolStr> {
+        self.propagate_header_names.to_owned()
     }
 
     /// Propagate filtered headers from original requests
-    pub fn propagate_headers(&mut self, headers: &HeaderMap<HeaderValue>) -> Result<()> {
+    pub fn propagate_headers(&mut self, headers: HeaderMap) -> Result<()> {
         self.propagate_headers.clear();
 
         if self.strategy == BackendStrategy::FastEdge {
             let server_name = headers
-                .get("Server_Name")
+                .get("server_name")
                 .and_then(|v| v.to_str().ok())
                 .ok_or(anyhow!("header Server_name is missing"))?;
-            self.propagate_headers
-                .push(("Host".to_string(), be_base_domain(server_name)));
+            self.propagate_headers.insert(
+                HeaderName::from_static("host"),
+                be_base_domain(server_name).parse()?,
+            );
         }
-
-        for header_name in self.propagate_header_names.iter() {
-            if let Some(value) = headers.get(header_name).and_then(|v| v.to_str().ok()) {
-                trace!("add original request header: {}={}", header_name, value);
-                self.propagate_headers
-                    .push((header_name.to_string(), value.to_string()));
+        let headers = headers.into_iter().filter(|(k, _)| {
+            if let Some(name) = k {
+                self.propagate_header_names.contains(&name.to_smolstr())
+            } else {
+                false
             }
-        }
+        });
+        self.propagate_headers.extend(headers);
+
         Ok(())
     }
 
-    fn make_request(&self, req: Request) -> Result<http::Request<hyper::Body>> {
+    fn propagate_headers_vec(&self) -> Vec<(String, String)> {
+        self.propagate_headers
+            .iter()
+            .filter_map(|(k, v)| v.to_str().ok().map(|v| (k.to_string(), v.to_string())))
+            .collect::<Vec<(String, String)>>()
+    }
+
+    fn make_request(&self, req: Request) -> Result<http::Request<Full<Bytes>>> {
         trace!("strategy: {:?}", self.strategy);
         let builder = match self.strategy {
             BackendStrategy::Direct => {
                 let mut headers = req.headers.into_iter().collect::<Vec<(String, String)>>();
-                headers.extend(self.propagate_headers.clone());
+                headers.extend(self.propagate_headers_vec());
                 // CLI has to set Host header from URL, if it is not set already by the request
                 if !headers
                     .iter()
-                    .any(|(k, _)| k.eq_ignore_ascii_case(HOST_HEADER_NAME))
+                    .any(|(k, _)| k.eq_ignore_ascii_case(header::HOST.as_str()))
                 {
                     if let Ok(uri) = req.uri.parse::<Uri>() {
                         if let Some(host) = uri.authority().map(|a| {
@@ -152,7 +169,7 @@ impl<C> Backend<C> {
                                 a.host().to_string()
                             }
                         }) {
-                            headers.push((HOST_HEADER_NAME.to_string(), host))
+                            headers.push((header::HOST.as_str().to_string(), host))
                         }
                     }
                 }
@@ -216,8 +233,12 @@ impl<C> Backend<C> {
                     })
                     .collect::<Vec<(String, String)>>();
 
-                headers.extend(backend_headers(&original_url, original_host));
-                headers.extend(self.propagate_headers.clone());
+                headers.push(("fastedge-hostname".to_string(), original_host));
+                headers.push((
+                    "fastedge-scheme".to_string(),
+                    original_url.scheme_str().unwrap_or("http").to_string(),
+                ));
+                headers.extend(self.propagate_headers_vec());
 
                 let host = canonical_host_name(&headers, &original_url)?;
                 let url = canonical_url(&original_url, &host, self.uri.path())?;
@@ -240,7 +261,7 @@ impl<C> Backend<C> {
         };
         debug!("request builder: {:?}", builder);
         let body = req.body.unwrap_or_default();
-        builder.body(hyper::Body::from(body)).map_err(Error::msg)
+        Ok(builder.body(Full::new(Bytes::from(body)))?)
     }
 }
 
@@ -280,7 +301,7 @@ where
             None
         };
 
-        let body_bytes = hyper::body::to_bytes(body).await?;
+        let body_bytes = body.collect().await?.to_bytes();
         let body = Some(body_bytes.to_vec());
 
         trace!(?status, ?headers, len = body_bytes.len(), "reply");
@@ -350,16 +371,6 @@ fn canonical_url(original_url: &Uri, canonical_host: &str, backend_path: &str) -
         .map_err(Error::msg)
 }
 
-fn backend_headers(original_url: &Uri, original_host: String) -> HeaderList {
-    vec![("Fastedge-Hostname".to_string(), original_host),(
-        "Fastedge-Scheme".to_string(),
-        original_url
-            .scheme_str()
-            .unwrap_or( "http")
-            .to_string(),
-    )]
-}
-
 impl FastEdgeConnector {
     pub fn new(backend: Uri) -> Self {
         let mut inner = HttpConnector::new();
@@ -385,54 +396,63 @@ impl Service<Uri> for FastEdgeConnector {
         Box::pin(async move {
             let conn = connect_fut
                 .await
-                .map(|inner| Connection { inner })
+                .map(|inner| Connection {
+                    inner: inner.into_inner(),
+                })
                 .map_err(Box::new)?;
             Ok(conn)
         })
     }
 }
 
-impl AsyncRead for Connection {
+impl hyper::rt::Read 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)
+        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 {
+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>> {
-        let this = self.project();
-        this.inner.poll_write(cx, buf)
+        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)
+        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)
+        tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
     }
 }
 
-impl hyper::client::connect::Connection for Connection {
-    fn connected(&self) -> hyper::client::connect::Connected {
-        hyper::client::connect::Connected::new()
+impl hyper_util::client::legacy::connect::Connection for Connection {
+    fn connected(&self) -> hyper_util::client::legacy::connect::Connected {
+        hyper_util::client::legacy::connect::Connected::new()
     }
 }