[xitca-web] update dep and db api. (#8365)

* update dep and db api.

* fix benchmark config

* fix dockerfile.

frameworks/Rust/xitca-web/Cargo.toml

@@ -29,7 +29,7 @@ pg = ["xitca-postgres"]
 # pg io_uring optional
 pg-iou = ["xitca-postgres/io-uring"]
 # pg orm optional
-pg-orm = ["diesel", "diesel-async", "tang-rs"]
+pg-orm = ["diesel", "diesel-async", "tang-rs", "futures-util"]
 # serde optional
 serde = ["dep:serde", "serde_json"]
 # web optional
@@ -55,8 +55,8 @@ xitca-web = { version = "0.1", features = ["json"], optional = true }
 xitca-postgres = { version = "0.1", features = ["single-thread"], optional = true }
 
 # diesel-pg orm optional
-diesel = { version = "2", default-features = false, features = ["i-implement-a-third-party-backend-and-opt-into-breaking-changes"], optional = true }
-diesel-async = { version = "0.2", default-features = false, features = ["postgres"], optional = true }
+diesel = { version = "2.1", default-features = false, features = ["i-implement-a-third-party-backend-and-opt-into-breaking-changes"], optional = true }
+diesel-async = { version = "0.3", default-features = false, features = ["postgres"], optional = true }
 tang-rs = { version = "0.2", optional = true }
 
 # serde optional
@@ -64,12 +64,15 @@ serde = { version = "1", optional = true }
 serde_json = { version = "1", optional = true }
 
 # template optional
-sailfish = { version = "0.6", optional = true }
+sailfish = { version = "0.8", default-features = false, features = ["derive", "perf-inline"], optional = true }
+
+# util optional
+futures-util = { version = "0.3", default-features = false, features = ["alloc"], optional = true }
 
 # stuff can not be used or not needed in wasi target
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
+futures-core = { version = "0.3", default-features = false }
 mimalloc = { version = "0.1", default-features = false }
-futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
 nanorand = { version = "0.7", default-features = false, features = ["tls"] }
 tokio = "1"
 
@@ -80,13 +83,13 @@ codegen-units = 1
 panic = "abort"
 
 [patch.crates-io]
-xitca-http = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-xitca-io = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-xitca-postgres = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-xitca-router = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-xitca-server = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-xitca-service = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-xitca-unsafe-collection = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-xitca-web = { git = "https://github.com/HFQR/xitca-web.git", rev = "3255e9ab0c585a18989bb6b8dd87b333ce662876" }
-
-mio = { git = "https://github.com/fakeshadow/mio.git", rev = "e506b0d87aa89e06c450a3991c491de35968cb12" }
+xitca-http = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+xitca-io = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+xitca-postgres = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+xitca-router = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+xitca-server = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+xitca-service = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+xitca-unsafe-collection = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+xitca-web = { git = "https://github.com/HFQR/xitca-web.git", rev = "c5b22169db1dde1a2411872cffa5b6bd7e14d2f4" }
+
+mio = { git = "https://github.com/fakeshadow/mio.git", rev = "eb67f6794edba8bc2e973ddef32e066b41ff812a" }

frameworks/Rust/xitca-web/benchmark_config.json

@@ -33,7 +33,7 @@
         "update_url": "/updates?q=",
         "port": 8080,
         "approach": "Realistic",
-        "classification": "Micro",
+        "classification": "Fullstack",
         "database": "Postgres",
         "framework": "xitca-web",
         "language": "Rust",

frameworks/Rust/xitca-web/rust-toolchain.toml

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "nightly-2023-05-04"
+channel = "nightly-2023-06-28"

frameworks/Rust/xitca-web/src/db.rs

@@ -1,6 +1,5 @@
-use std::{cell::RefCell, collections::HashMap, fmt::Write, future::Future};
+use std::{cell::RefCell, collections::HashMap, fmt::Write};
 
-use futures_util::future::{try_join, try_join_all, TryFutureExt};
 use xitca_postgres::{statement::Statement, AsyncIterator, Postgres};
 use xitca_unsafe_collection::no_hash::NoHashBuilder;
 
@@ -82,7 +81,8 @@ pub async fn create(config: &str) -> HandleResult<Client> {
 }
 
 impl Client {
-    async fn query_one_world(&self, id: i32) -> HandleResult<World> {
+    pub async fn get_world(&self) -> HandleResult<World> {
+        let id = self.rng.borrow_mut().gen_id();
         self.client
             .query_raw(&self.world, [id])
             .await?
@@ -93,15 +93,25 @@ impl Client {
             .map_err(Into::into)
     }
 
-    pub fn get_world(&self) -> impl Future<Output = HandleResult<World>> + '_ {
-        let id = self.rng.borrow_mut().gen_id();
-        self.query_one_world(id)
-    }
+    pub async fn get_worlds(&self, num: u16) -> HandleResult<Vec<World>> {
+        let mut pipe = self.client.pipeline();
+
+        {
+            let mut rng = self.rng.borrow_mut();
+            (0..num).try_for_each(|_| pipe.query_raw(&self.world, [rng.gen_id()]))?;
+        }
+
+        let mut worlds = Vec::new();
+        worlds.reserve(num as usize);
+
+        let mut res = pipe.run().await?;
+        while let Some(mut item) = res.next().await.transpose()? {
+            while let Some(row) = item.next().await.transpose()? {
+                worlds.push(World::new(row.get_raw(0), row.get_raw(1)))
+            }
+        }
 
-    pub fn get_worlds(&self, num: u16) -> impl Future<Output = HandleResult<Vec<World>>> + '_ {
-        let mut rng = self.rng.borrow_mut();
-        let gets = (0..num).map(|_| self.query_one_world(rng.gen_id()));
-        try_join_all(gets)
+        Ok(worlds)
     }
 
     pub async fn update(&self, num: u16) -> HandleResult<Vec<World>> {
@@ -110,41 +120,45 @@ impl Client {
         let mut params = Vec::new();
         params.reserve(len * 3);
 
-        let gets = {
+        let mut pipe = self.client.pipeline();
+
+        {
             let mut rng = self.rng.borrow_mut();
-            let gets = (0..num).map(|_| {
+            (0..num).try_for_each(|_| {
                 let w_id = rng.gen_id();
                 let r_id = rng.gen_id();
-                params.push(w_id);
-                params.push(r_id);
-                self.query_one_world(w_id).map_ok(move |mut world| {
-                    world.randomnumber = r_id;
-                    world
-                })
-            });
-            try_join_all(gets)
-        };
+                params.extend([w_id, r_id]);
+                pipe.query_raw(&self.world, [w_id])
+            })?;
+        }
 
         params.extend_from_within(..len);
-
         let st = self.updates.get(&num).unwrap();
-        let update = self.client.execute_raw(st, &params).map_err(Into::into);
+        pipe.query_raw(st, &params)?;
+
+        let mut worlds = Vec::new();
+        worlds.reserve(len);
+        let mut r_ids = params.into_iter().skip(1).step_by(2);
+
+        let mut res = pipe.run().await?;
+        while let Some(mut item) = res.next().await.transpose()? {
+            while let Some(row) = item.next().await.transpose()? {
+                let r_id = r_ids.next().unwrap();
+                worlds.push(World::new(row.get_raw(0), r_id))
+            }
+        }
 
-        try_join(gets, update).await.map(|(world, _)| world)
+        Ok(worlds)
     }
 
     pub async fn tell_fortune(&self) -> HandleResult<Fortunes> {
         let mut items = Vec::with_capacity(32);
-
         items.push(Fortune::new(0, "Additional fortune added at request time."));
 
         let mut stream = self.client.query_raw::<[i32; 0]>(&self.fortune, []).await?;
-
-        while let Some(row) = stream.next().await {
-            let row = row?;
+        while let Some(row) = stream.next().await.transpose()? {
             items.push(Fortune::new(row.get_raw(0), row.get_raw::<String>(1)));
         }
-
         items.sort_by(|it, next| it.message.cmp(&next.message));
 
         Ok(Fortunes::new(items))

frameworks/Rust/xitca-web/src/db_diesel.rs

@@ -2,7 +2,7 @@ use std::{cell::RefCell, error::Error, fmt, future::Future, io, time::Duration};
 
 use diesel::prelude::{ConnectionError, ExpressionMethods, QueryDsl};
 use diesel_async::{AsyncConnection, AsyncPgConnection, RunQueryDsl};
-use futures_util::stream::{FuturesUnordered, TryStreamExt};
+use futures_util::future::try_join_all;
 use tang_rs::{Manager, ManagerFuture, ManagerTimeout, Pool};
 use tokio::time::{sleep, Sleep};
 
@@ -132,23 +132,19 @@ impl DieselPool {
     pub async fn get_worlds(&self, num: u16) -> DbResult<Vec<World>> {
         use crate::schema::world::dsl::*;
 
-        let worlds = {
+        {
             let mut conn = self.pool.get().await?;
-
             let mut rng = self.rng.borrow_mut();
-            (0..num)
-                .map(|_| {
-                    let w_id = rng.gen_id();
-                    let fut = world.filter(id.eq(w_id)).load::<World>(&mut *conn);
-                    async {
-                        let w = fut.await?.pop().unwrap();
-                        Ok(w)
-                    }
-                })
-                .collect::<FuturesUnordered<_>>()
-        };
-
-        worlds.try_collect().await
+            try_join_all((0..num).map(|_| {
+                let w_id = rng.gen_id();
+                let fut = world.filter(id.eq(w_id)).load::<World>(&mut *conn);
+                async {
+                    let w = fut.await?.pop().unwrap();
+                    Ok(w)
+                }
+            }))
+        }
+        .await
     }
 
     pub async fn update(&self, num: u16) -> DbResult<Vec<World>> {
@@ -158,20 +154,17 @@ impl DieselPool {
 
         let mut worlds = {
             let mut rng = self.rng.borrow_mut();
-            (0..num)
-                .map(|_| {
-                    let w_id = rng.gen_id();
-                    let new_id = rng.gen_id();
-                    let fut = world.filter(id.eq(w_id)).load::<World>(&mut *conn);
-                    async move {
-                        let mut w = fut.await?.pop().unwrap();
-                        w.randomnumber = new_id;
-                        DbResult::Ok(w)
-                    }
-                })
-                .collect::<FuturesUnordered<_>>()
+            try_join_all((0..num).map(|_| {
+                let w_id = rng.gen_id();
+                let new_id = rng.gen_id();
+                let fut = world.filter(id.eq(w_id)).load::<World>(&mut *conn);
+                async move {
+                    let mut w = fut.await?.pop().unwrap();
+                    w.randomnumber = new_id;
+                    DbResult::Ok(w)
+                }
+            }))
         }
-        .try_collect::<Vec<_>>()
         .await?;
 
         worlds.sort_by_key(|w| w.id);

frameworks/Rust/xitca-web/src/main_iou.rs

@@ -11,9 +11,16 @@ mod db;
 mod ser;
 mod util;
 
-use std::{cell::RefCell, convert::Infallible, fmt, future::Future, io};
+use std::{
+    cell::RefCell,
+    convert::Infallible,
+    fmt,
+    future::{poll_fn, Future},
+    io,
+    pin::pin,
+};
 
-use futures_util::stream::StreamExt;
+use futures_core::stream::Stream;
 use xitca_http::{
     body::Once,
     date::DateTimeService,
@@ -184,9 +191,13 @@ where
 
                 while let Some((req, _)) = ctx.decode_head::<{ usize::MAX }>(&mut read_buf).unwrap()
                 {
-                    let (parts, mut body) = self.service.call(req).await.unwrap().into_parts();
+                    let (parts, body) = self.service.call(req).await.unwrap().into_parts();
                     let mut encoder = ctx.encode_head(parts, &body, &mut write_buf).unwrap();
-                    let chunk = body.next().now_or_panic().unwrap().unwrap();
+                    let mut body = pin!(body);
+                    let chunk = poll_fn(|cx| body.as_mut().poll_next(cx))
+                        .now_or_panic()
+                        .unwrap()
+                        .unwrap();
                     encoder.encode(chunk, &mut write_buf);
                     encoder.encode_eof(&mut write_buf);
                 }

frameworks/Rust/xitca-web/xitca-web-wasm.dockerfile

@@ -1,11 +1,11 @@
 FROM rust:1.67 AS compile
 
-ARG WASMTIME_VERSION=8.0.1
+ARG WASMTIME_VERSION=11.0.1
 
 WORKDIR /tmp
 COPY / ./
 RUN curl -LSs "https://github.com/bytecodealliance/wasmtime/releases/download/v${WASMTIME_VERSION}/wasmtime-v${WASMTIME_VERSION}-$(uname -m)-linux.tar.xz" | \
-tar --strip-components=2 -Jx && \
+tar --strip-components=1 -Jx && \
 rustup target add wasm32-wasi && \
 cargo build --bin xitca-web-wasm --features serde,web --release --target wasm32-wasi