clean up module layout

Author: fakeshadow

frameworks/Rust/xitca-web/Cargo.lock

@@ -10,7 +10,7 @@ use super::{
     util::{DB_URL, HandleResult},
 };
 
-use db_util::{FORTUNE_STMT, Shared, UPDATE_BATCH_STMT, WORLD_STMT, not_found};
+use db_util::{FORTUNE_STMT, Shared, UPDATE_STMT, WORLD_STMT, not_found};
 
 pub struct Client {
     pool: Pool,
@@ -60,7 +60,7 @@ impl Client {
     pub async fn update(&self, num: u16) -> HandleResult<Vec<World>> {
         let mut conn = self.pool.get().await?;
         let world_stmt = WORLD_STMT.execute(&mut conn).await?;
-        let update_stmt = UPDATE_BATCH_STMT.execute(&mut conn).await?;
+        let update_stmt = UPDATE_STMT.execute(&mut conn).await?;
 
         let (mut res, worlds) = {
             let (ref mut rng, ref mut buf) = *self.shared.borrow_mut();

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

@@ -1,10 +1,3 @@
-#[path = "./db_util.rs"]
-mod db_util;
-
-use core::cell::RefCell;
-
-use std::io;
-
 use diesel::prelude::*;
 use diesel_async::{
     RunQueryDsl,
@@ -18,31 +11,29 @@ use crate::{
     util::{DB_URL, HandleResult, Rand},
 };
 
-use db_util::update_query_from_ids;
-
 pub struct Pool {
     pool: bb8::Pool<AsyncPgConnection>,
-    rng: RefCell<Rand>,
+    rng: core::cell::RefCell<Rand>,
 }
 
-pub async fn create() -> io::Result<Pool> {
-    bb8::Pool::builder()
-        .max_size(1)
-        .min_idle(Some(1))
-        .test_on_check_out(false)
-        .build(AsyncDieselConnectionManager::new(DB_URL))
-        .await
-        .map_err(io::Error::other)
-        .map(|pool| Pool {
+impl Pool {
+    pub async fn create() -> HandleResult<Self> {
+        let pool = bb8::Pool::builder()
+            .max_size(1)
+            .min_idle(Some(1))
+            .test_on_check_out(false)
+            .build(AsyncDieselConnectionManager::new(DB_URL))
+            .await?;
+
+        Ok(Self {
             pool,
-            rng: RefCell::new(Rand::default()),
+            rng: Default::default(),
         })
-}
+    }
 
-impl Pool {
     pub async fn get_world(&self) -> HandleResult<World> {
         {
-            use crate::schema::world::dsl::*;
+            use schema::world::dsl::*;
 
             let w_id = self.rng.borrow_mut().gen_id();
             let mut conn = self.pool.get().await?;
@@ -53,45 +44,44 @@ impl Pool {
 
     pub async fn get_worlds(&self, num: u16) -> HandleResult<Vec<World>> {
         {
-            use crate::schema::world::dsl::*;
+            use schema::world::dsl::*;
 
             let mut conn = self.pool.get().await?;
-            let mut rng = self.rng.borrow_mut();
-
-            core::iter::repeat_with(move || {
-                let w_id = rng.gen_id();
-                world.filter(id.eq(w_id)).first(&mut conn).map_err(Into::into)
-            })
-            .take(num as _)
-            .collect::<TryJoinAll<_>>()
+            self.rng
+                .borrow_mut()
+                .gen_multi()
+                .take(num as _)
+                .map(|w_id| world.filter(id.eq(w_id)).first(&mut conn).map_err(Into::into))
+                .collect::<TryJoinAll<_>>()
         }
         .await
     }
 
     pub async fn update(&self, num: u16) -> HandleResult<Vec<World>> {
         {
-            use crate::schema::world::dsl::*;
+            use schema::world::dsl::*;
 
             let mut conn = self.pool.get().await?;
             let mut rng = self.rng.borrow_mut();
             let mut params = Vec::with_capacity(num as _);
 
-            let get = core::iter::repeat_with(|| {
-                let w_id = rng.gen_id();
-                let rng = rng.gen_id();
+            let get = rng
+                .clone()
+                .gen_multi()
+                .take(num as _)
+                .zip(rng.gen_multi())
+                .map(|(w_id, rng)| {
+                    let get = world.filter(id.eq(w_id)).first::<World>(&mut conn);
 
-                let get = world.filter(id.eq(w_id)).first::<World>(&mut conn);
+                    params.push((w_id, rng));
 
-                params.push((w_id, rng));
-
-                async move {
-                    let mut w = get.await?;
-                    w.randomnumber = rng;
-                    HandleResult::Ok(w)
-                }
-            })
-            .take(num as _)
-            .collect::<TryJoinAll<_>>();
+                    async move {
+                        let mut w = get.await?;
+                        w.randomnumber = rng;
+                        HandleResult::Ok(w)
+                    }
+                })
+                .collect::<TryJoinAll<_>>();
 
             let sql = update_query_from_ids(params);
             let update = diesel::sql_query(sql).execute(&mut conn).map_err(Into::into);
@@ -104,7 +94,7 @@ impl Pool {
 
     pub async fn tell_fortune(&self) -> HandleResult<Fortunes> {
         {
-            use crate::schema::fortune::dsl::*;
+            use schema::fortune::dsl::*;
 
             let mut conn = self.pool.get().await?;
             fortune.load(&mut conn).map_err(Into::into)
@@ -113,3 +103,43 @@ impl Pool {
         .map(Fortunes::new)
     }
 }
+
+mod schema {
+    diesel::table! {
+        world (id) {
+            id -> Integer,
+            randomnumber -> Integer,
+        }
+    }
+
+    diesel::table! {
+        fortune (id) {
+            id -> Integer,
+            message -> Text,
+        }
+    }
+}
+
+// diesel does not support high level bulk update api. use raw sql to bypass the limitation.
+// relate discussion: https://github.com/diesel-rs/diesel/discussions/2879
+fn update_query_from_ids(mut rngs: Vec<(i32, i32)>) -> String {
+    rngs.sort_by(|(a, _), (b, _)| a.cmp(b));
+
+    const PREFIX: &str = "UPDATE world SET randomNumber=w.r FROM (VALUES ";
+    const SUFFIX: &str = ") AS w (i,r) WHERE world.id=w.i";
+
+    let mut query = String::from(PREFIX);
+
+    use core::fmt::Write;
+    rngs.iter().for_each(|(w_id, num)| {
+        write!(query, "({}::int,{}::int),", w_id, num).unwrap();
+    });
+
+    if query.ends_with(',') {
+        query.pop();
+    }
+
+    query.push_str(SUFFIX);
+
+    query
+}

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

@@ -12,22 +12,22 @@ pub struct Pool {
     rng: core::cell::RefCell<Rand>,
 }
 
-pub async fn create() -> HandleResult<Pool> {
-    let conn = xitca_postgres_toasty::PostgreSQL::connect(DB_URL).await?;
-
-    let db = Db::builder()
-        .register::<World>()
-        .register::<Fortune>()
-        .build(conn)
-        .await?;
-
-    Ok(Pool {
-        db,
-        rng: Default::default(),
-    })
-}
-
 impl Pool {
+    pub async fn create() -> HandleResult<Self> {
+        let conn = xitca_postgres_toasty::PostgreSQL::connect(DB_URL).await?;
+
+        let db = Db::builder()
+            .register::<World>()
+            .register::<Fortune>()
+            .build(conn)
+            .await?;
+
+        Ok(Self {
+            db,
+            rng: Default::default(),
+        })
+    }
+
     pub async fn get_world(&self) -> HandleResult<World> {
         let id = self.rng.borrow_mut().gen_id();
         World::get_by_id(&self.db, id).await.map_err(Into::into)

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

@@ -26,10 +26,10 @@ pub struct Client {
 pub async fn create() -> HandleResult<Client> {
     let (cli, mut drv) = xitca_postgres::Postgres::new(DB_URL).connect().await?;
 
-    tokio::task::spawn(tokio::task::unconstrained(async move {
+    tokio::task::spawn(async move {
         while drv.try_next().await?.is_some() {}
         HandleResult::Ok(())
-    }));
+    });
 
     let world = WORLD_STMT.execute(&cli).await?.leak();
     let fortune = FORTUNE_STMT.execute(&cli).await?.leak();
@@ -53,16 +53,16 @@ impl Client {
     }
 
     pub async fn get_worlds(&self, num: u16) -> HandleResult<Vec<World>> {
-        let len = num as usize;
-
         let mut res = {
             let (ref mut rng, ref mut buf) = *self.shared.borrow_mut();
-            let mut pipe = Pipeline::with_capacity_from_buf(len, buf);
-            (0..num).try_for_each(|_| self.world.bind([rng.gen_id()]).query(&mut pipe))?;
+            let mut pipe = Pipeline::with_capacity_from_buf(num as _, buf);
+            rng.gen_multi()
+                .take(num as _)
+                .try_for_each(|id| self.world.bind([id]).query(&mut pipe))?;
             pipe.query(&self.cli)?
         };
 
-        let mut worlds = Vec::with_capacity(len);
+        let mut worlds = Vec::with_capacity(num as _);
 
         while let Some(mut item) = res.try_next().await? {
             while let Some(row) = item.try_next().await? {
@@ -78,23 +78,21 @@ impl Client {
 
         let (mut res, worlds) = {
             let (ref mut rng, ref mut buf) = *self.shared.borrow_mut();
-            // unrealistic as all queries are sent with only one sync point.
-            let mut pipe = Pipeline::unsync_with_capacity_from_buf(len + 1, buf);
-
-            let (mut params, worlds) = core::iter::repeat_with(|| {
-                let id = rng.gen_id();
-                let rand = rng.gen_id();
-                self.world.bind([id]).query(&mut pipe)?;
-                HandleResult::Ok(((id, rand), World::new(id, rand)))
-            })
-            .take(len)
-            .collect::<HandleResult<(Vec<_>, Vec<_>)>>()?;
-
-            params.sort();
-
-            params
-                .into_iter()
-                .try_for_each(|(id, rng)| self.update.bind([rng, id]).query(&mut pipe))?;
+            let mut pipe = Pipeline::with_capacity_from_buf(len + 1, buf);
+
+            let mut ids = rng.gen_multi().take(num as _).collect::<Vec<_>>();
+            ids.sort();
+
+            let (rngs, worlds) = ids
+                .iter()
+                .cloned()
+                .zip(rng.gen_multi())
+                .map(|(id, rand)| {
+                    self.world.bind([id]).query(&mut pipe)?;
+                    HandleResult::Ok((rand, World::new(id, rand)))
+                })
+                .collect::<HandleResult<(Vec<_>, Vec<_>)>>()?;
+            self.update.bind([&ids, &rngs]).query(&mut pipe)?;
 
             (pipe.query(&self.cli)?, worlds)
         };

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

@@ -1,28 +1,3 @@
-#[cfg(feature = "diesel")]
-// diesel does not support high level bulk update api. use raw sql to bypass the limitation.
-// relate discussion: https://github.com/diesel-rs/diesel/discussions/2879
-pub fn update_query_from_ids(mut rngs: Vec<(i32, i32)>) -> String {
-    rngs.sort_by(|(a, _), (b, _)| a.cmp(b));
-
-    const PREFIX: &str = "UPDATE world SET randomNumber=w.r FROM (VALUES ";
-    const SUFFIX: &str = ") AS w (i,r) WHERE world.id=w.i";
-
-    let mut query = String::from(PREFIX);
-
-    use core::fmt::Write;
-    rngs.iter().for_each(|(w_id, num)| {
-        write!(query, "({}::int,{}::int),", w_id, num).unwrap();
-    });
-
-    if query.ends_with(',') {
-        query.pop();
-    }
-
-    query.push_str(SUFFIX);
-
-    query
-}
-
 #[cfg(feature = "pg")]
 pub use pg::*;
 
@@ -44,10 +19,6 @@ pub mod pg {
     pub const WORLD_STMT: StatementNamed =
         Statement::named("SELECT id,randomnumber FROM world WHERE id=$1", &[Type::INT4]);
     pub const UPDATE_STMT: StatementNamed = Statement::named(
-        "UPDATE world SET randomnumber=$1 WHERE id=$2",
-        &[Type::INT4, Type::INT4],
-    );
-    pub const UPDATE_BATCH_STMT: StatementNamed = Statement::named(
         "UPDATE world SET randomnumber=w.r FROM (SELECT unnest($1) as i,unnest($2) as r) w WHERE world.id=w.i",
         &[Type::INT4_ARRAY, Type::INT4_ARRAY],
     );