ntex: Upgrade to ntex-2.0 (#9185)

Author: fafhrd91

frameworks/Rust/ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "1.0.0"
+version = "2.0.0"
 edition = "2018"
 
 [[bin]]
@@ -37,22 +37,23 @@ tokio = ["ntex/tokio"]
 async-std = ["ntex/async-std"]
 
 [dependencies]
-ntex = "1.0.0"
+ntex = "=2.0.3"
 ntex-bytes = { version = "0.1.21", features=["simd"] }
 mimalloc = { version = "0.1.25", default-features = false }
 snmalloc-rs = { version = "0.3.3", features = ["native-cpu"] }
 yarte = { version = "0.15", features = ["bytes-buf", "json"] }
 buf-min = { version = "0.7", features = ["ntex-bytes"] }
-env_logger = "0.10"
+env_logger = "0.11"
 nanorand = { version = "0.7", default-features = false, features = ["std", "wyrand", "tls"] }
 atoi = "2.0"
 num_cpus = "1.16"
-smallvec = "1.11"
+smallvec = "1.13"
+futures = "0.3"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 log = { version = "0.4", features = ["release_max_level_off"] }
 tok_io = {version = "1", package = "tokio" }
-tokio-postgres = { git="https://github.com/fafhrd91/postgres.git", branch="ntex-1.0" }
+tokio-postgres = { git="https://github.com/fafhrd91/postgres.git", branch="ntex-2" }
 
 [profile.release]
 opt-level = 3

frameworks/Rust/ntex/src/db.rs

@@ -1,10 +1,11 @@
 use std::{cell::RefCell, fmt::Write as FmtWrite};
 
+use futures::stream::{futures_unordered::FuturesUnordered, StreamExt};
 use nanorand::{Rng, WyRand};
 use ntex::util::{BufMut, Bytes, BytesMut};
 use smallvec::SmallVec;
 use tokio_postgres::types::ToSql;
-use tokio_postgres::{connect, Client, Statement};
+use tokio_postgres::{connect, Client, Row, Statement};
 use yarte::{ywrite_html, Serialize};
 
 use super::utils;
@@ -82,7 +83,7 @@ impl PgConnection {
         let row = self.cl.query_one(&self.world, &[&random_id]).await.unwrap();
 
         let mut body = self.buf.borrow_mut();
-        utils::reserve(&mut body);
+        utils::reserve(&mut body, 256);
         World {
             id: row.get(0),
             randomnumber: row.get(1),
@@ -91,25 +92,28 @@ impl PgConnection {
         body.split().freeze()
     }
 
+    async fn get_one_world(&self, id: i32) -> Row {
+        self.cl.query_one(&self.world, &[&id]).await.unwrap()
+    }
+
     pub async fn get_worlds(&self, num: usize) -> Bytes {
         let mut rng = self.rng.clone();
-        let mut queries = SmallVec::<[_; 32]>::new();
+        let mut queries = FuturesUnordered::new();
         (0..num).for_each(|_| {
             let w_id = (rng.generate::<u32>() % 10_000 + 1) as i32;
-            queries.push(self.cl.query_one(&self.world, &[&w_id]));
+            queries.push(self.get_one_world(w_id))
         });
 
         let mut worlds = SmallVec::<[_; 32]>::new();
-        for fut in queries {
-            let row = fut.await.unwrap();
+        while let Some(row) = queries.next().await {
             worlds.push(World {
                 id: row.get(0),
                 randomnumber: row.get(1),
             })
         }
 
         let mut body = self.buf.borrow_mut();
-        utils::reserve(&mut body);
+        utils::reserve(&mut body, 8 * 1024);
         body.put_u8(b'[');
         worlds.iter().for_each(|w| {
             w.to_bytes_mut(&mut *body);
@@ -121,16 +125,15 @@ impl PgConnection {
     }
 
     pub async fn update(&self, num: usize) -> Bytes {
-        let mut rng = nanorand::tls_rng();
-        let mut queries = SmallVec::<[_; 32]>::new();
+        let mut rng = self.rng.clone();
+        let mut queries = FuturesUnordered::new();
         (0..num).for_each(|_| {
             let w_id = (rng.generate::<u32>() % 10_000 + 1) as i32;
-            queries.push(self.cl.query_one(&self.world, &[&w_id]));
+            queries.push(self.get_one_world(w_id))
         });
 
         let mut worlds = SmallVec::<[_; 32]>::new();
-        for fut in queries.into_iter() {
-            let row = fut.await.unwrap();
+        while let Some(row) = queries.next().await {
             worlds.push(World {
                 id: row.get(0),
                 randomnumber: (rng.generate::<u32>() % 10_000 + 1) as i32,
@@ -148,7 +151,7 @@ impl PgConnection {
         let _ = self.cl.query(&self.updates[num - 1], &params).await;
 
         let mut body = self.buf.borrow_mut();
-        utils::reserve(&mut body);
+        utils::reserve(&mut body, 8 * 1024);
         body.put_u8(b'[');
         worlds.iter().for_each(|w| {
             w.to_bytes_mut(&mut *body);
@@ -174,7 +177,7 @@ impl PgConnection {
         fortunes.sort_by(|it, next| it.message.cmp(next.message));
 
         let mut body = std::mem::replace(&mut *self.buf.borrow_mut(), BytesMut::new());
-        utils::reserve(&mut body);
+        utils::reserve(&mut body, 8 * 1024);
         ywrite_html!(body, "{{> fortune }}");
         let result = body.split().freeze();
         let _ = std::mem::replace(&mut *self.buf.borrow_mut(), body);

frameworks/Rust/ntex/src/main_plt.rs

@@ -35,7 +35,7 @@ impl Future for App {
                 Ok((req, _)) => {
                     let _ = this.io.with_write_buf(|buf| {
                         buf.with_bytes_mut(|buf| {
-                            utils::reserve(buf);
+                            utils::reserve(buf, 2 * 1024);
                             match req.path() {
                                 "/json" => {
                                     buf.extend_from_slice(JSON);

frameworks/Rust/ntex/src/utils.rs

@@ -11,7 +11,6 @@ pub const HDR_HTML_CONTENT_TYPE: HeaderValue =
     HeaderValue::from_static("text/html; charset=utf-8");
 pub const BODY_PLAIN_TEXT: Bytes = Bytes::from_static(b"Hello, World!");
 
-const LW: usize = 1024;
 const HW: usize = 128 * 1024;
 pub const SIZE: usize = 27;
 
@@ -25,9 +24,9 @@ pub fn get_query_param(query: Option<&str>) -> usize {
     cmp::min(500, cmp::max(1, q) as usize)
 }
 
-pub fn reserve(buf: &mut BytesMut) {
+pub fn reserve(buf: &mut BytesMut, lw: usize) {
     let remaining = buf.remaining_mut();
-    if remaining < LW {
+    if remaining < lw {
         buf.reserve(HW);
     }
 }