[xitca-web] join futures (TechEmpower#8239)

Author: fakeshadow

frameworks/Rust/xitca-web/Cargo.lock

@@ -35,7 +35,7 @@ web = ["xitca-web"]
 # template optional
 template = ["sailfish"]
 # io-uring optional
-io-uring = ["tokio-uring", "xitca-http/io-uring", "xitca-server/io-uring"]
+io-uring = ["xitca-http/io-uring", "xitca-server/io-uring"]
 
 [dependencies]
 xitca-http = { version = "0.1", features = ["util-service"] }
@@ -61,9 +61,6 @@ tang-rs = { version = "0.2", optional = true }
 serde = { version = "1", optional = true }
 serde_json = { version = "1", optional = true }
 
-# io-uring optional
-tokio-uring = { version = "0.4", features = ["bytes"], optional = true }
-
 # template optional
 sailfish = { version = "0.6", optional = true }
 
@@ -90,4 +87,4 @@ xitca-service = { git = "https://github.com/HFQR/xitca-web.git", rev = "061a61ad
 xitca-unsafe-collection = { git = "https://github.com/HFQR/xitca-web.git", rev = "061a61adc33205a7d7a124cee5665522a3df4d59" }
 xitca-web = { git = "https://github.com/HFQR/xitca-web.git", rev = "061a61adc33205a7d7a124cee5665522a3df4d59" }
 
-mio = { git = "https://github.com/fakeshadow/mio.git", rev = "2e80aa89ee94a0e2d343331b8df106d7321148b9" }
+mio = { git = "https://github.com/fakeshadow/mio.git", rev = "e506b0d87aa89e06c450a3991c491de35968cb12" }

frameworks/Rust/xitca-web/Cargo.toml

@@ -5,7 +5,7 @@ use std::{
     future::{Future, IntoFuture},
 };
 
-use futures_util::stream::{FuturesUnordered, TryStreamExt};
+use futures_util::future::{try_join, try_join_all, TryFutureExt};
 use xitca_postgres::{statement::Statement, AsyncIterator, Postgres};
 use xitca_unsafe_collection::no_hash::NoHashBuilder;
 
@@ -94,46 +94,37 @@ impl Client {
 
     pub fn get_worlds(&self, num: u16) -> impl Future<Output = HandleResult<Vec<World>>> + '_ {
         let mut rng = self.rng.borrow_mut();
-        (0..num)
-            .map(|_| {
-                let id = rng.gen_id();
-                self.query_one_world(id)
-            })
-            .collect::<FuturesUnordered<_>>()
-            .try_collect()
+        let gets = (0..num).map(|_| self.query_one_world(rng.gen_id()));
+        try_join_all(gets)
     }
 
     pub async fn update(&self, num: u16) -> HandleResult<Vec<World>> {
-        let worlds = {
+        let len = num as usize;
+
+        let mut params = Vec::new();
+        params.reserve(len * 3);
+
+        let gets = {
             let mut rng = self.rng.borrow_mut();
-            (0..num)
-                .map(|_| {
-                    let id = rng.gen_id();
-                    let w_id = rng.gen_id();
-                    async move {
-                        self.query_one_world(w_id).await.map(|mut world| {
-                            world.randomnumber = id;
-                            world
-                        })
-                    }
+            let gets = (0..num).map(|_| {
+                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
                 })
-                .collect::<FuturesUnordered<_>>()
+            });
+            try_join_all(gets)
         };
 
-        let worlds = worlds.try_collect::<Vec<_>>().await?;
-
-        let params = worlds
-            .iter()
-            .map(|w| [w.id, w.randomnumber])
-            .flatten()
-            .chain(worlds.iter().map(|w| w.id))
-            .collect::<Vec<_>>();
+        params.extend_from_within(..len);
 
         let st = self.updates.get(&num).unwrap();
+        let update = self.client.execute_raw(st, &params).map_err(Into::into);
 
-        self.client.execute_raw(st, &params).await?;
-
-        Ok(worlds)
+        try_join(gets, update).await.map(|(world, _)| world)
     }
 
     pub async fn tell_fortune(&self) -> HandleResult<Fortunes> {

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

@@ -36,7 +36,7 @@ use xitca_http::{
 use xitca_io::{
     bytes::{Bytes, BytesMut},
     io_uring::IoBuf,
-    net::TcpStream,
+    net::{io_uring::TcpStream as IOUTcpStream, TcpStream},
 };
 use xitca_service::{fn_service, middleware::UncheckedReady, Service, ServiceExt};
 
@@ -173,7 +173,7 @@ where
             let mut write_buf = BytesMut::with_capacity(4096);
 
             let std = stream.into_std()?;
-            let stream = tokio_uring::net::TcpStream::from_std(std);
+            let stream = IOUTcpStream::from_std(std);
 
             loop {
                 let len = read_buf.len();