[xitca-web] strict thread-per-core (#9399)

* use core affinity for static thread per core

* use strict thread per core pattern

* dep update

* dep update

frameworks/Rust/xitca-web/Cargo.lock

@@ -66,9 +66,9 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
 
 [[package]]
 name = "bb8"
-version = "0.8.5"
+version = "0.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b10cf871f3ff2ce56432fddc2615ac7acc3aa22ca321f8fea800846fbb32f188"
+checksum = "d89aabfae550a5c44b43ab941844ffcd2e993cb6900b342debf59e9ea74acdb8"
 dependencies = [
  "async-trait",
  "futures-util",
@@ -111,15 +111,15 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
 [[package]]
 name = "bytes"
-version = "1.7.2"
+version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "428d9aa8fbc0670b7b8d6030a7fadd0f86151cae55e4dbbece15f3780a3dfaf3"
+checksum = "9ac0150caa2ae65ca5bd83f25c7de183dea78d4d366469f148435e2acfbad0da"
 
 [[package]]
 name = "cc"
-version = "1.1.30"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b16803a61b81d9eabb7eae2588776c4c1e584b738ede45fdbb4c972cec1e9945"
+checksum = "fd9de9f2205d5ef3fd67e685b0df337994ddd4495e2a28d185500d0e1edfea47"
 dependencies = [
  "shlex",
 ]
@@ -130,11 +130,22 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
+[[package]]
+name = "core_affinity"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "622892f5635ce1fc38c8f16dfc938553ed64af482edb5e150bf4caedbfcb2304"
+dependencies = [
+ "libc",
+ "num_cpus",
+ "winapi",
+]
+
 [[package]]
 name = "cpufeatures"
-version = "0.2.14"
+version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "608697df725056feaccfa42cffdaeeec3fccc4ffc38358ecd19b243e716a78e0"
+checksum = "0ca741a962e1b0bff6d724a1a0958b686406e853bb14061f218562e1896f95e6"
 dependencies = [
  "libc",
 ]
@@ -429,9 +440,9 @@ dependencies = [
 
 [[package]]
 name = "itoa"
-version = "1.0.11"
+version = "1.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "49f1f14873335454500d59611f1cf4a4b0f786f9ac11f4312a78e4cf2566695b"
+checksum = "7a73e9fe3c49d7afb2ace819fa181a287ce54a0983eda4e0eb05c22f82ffe534"
 
 [[package]]
 name = "itoap"
@@ -450,9 +461,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.159"
+version = "0.2.164"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "561d97a539a36e26a9a5fad1ea11a3039a67714694aaa379433e580854bc3dc5"
+checksum = "433bfe06b8c75da9b2e3fbea6e5329ff87748f0b144ef75306e674c3f6f7c13f"
 
 [[package]]
 name = "libmimalloc-sys"
@@ -534,6 +545,16 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "num_cpus"
+version = "1.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43"
+dependencies = [
+ "hermit-abi",
+ "libc",
+]
+
 [[package]]
 name = "object"
 version = "0.36.5"
@@ -598,9 +619,9 @@ dependencies = [
 
 [[package]]
 name = "pin-project-lite"
-version = "0.2.14"
+version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bda66fc9667c18cb2758a2ac84d1167245054bcf85d5d1aaa6923f45801bdd02"
+checksum = "915a1e146535de9163f3987b8944ed8cf49a18bb0056bcebcdcece385cece4ff"
 
 [[package]]
 name = "pin-utils"
@@ -657,9 +678,9 @@ dependencies = [
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.87"
+version = "1.0.89"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b3e4daa0dcf6feba26f985457cdf104d4b4256fc5a09547140f3631bb076b19a"
+checksum = "f139b0662de085916d1fb67d2b4169d1addddda1919e696f3252b740b629986e"
 dependencies = [
  "unicode-ident",
 ]
@@ -757,12 +778,11 @@ dependencies = [
 
 [[package]]
 name = "scoped-futures"
-version = "0.1.3"
+version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b1473e24c637950c9bd38763220bea91ec3e095a89f672bbd7a10d03e77ba467"
+checksum = "1b24aae2d0636530f359e9d5ef0c04669d11c5e756699b27a6a6d845d8329091"
 dependencies = [
- "cfg-if",
- "pin-utils",
+ "pin-project-lite",
 ]
 
 [[package]]
@@ -773,18 +793,18 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
 [[package]]
 name = "serde"
-version = "1.0.210"
+version = "1.0.215"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8e3592472072e6e22e0a54d5904d9febf8508f65fb8552499a1abc7d1078c3a"
+checksum = "6513c1ad0b11a9376da888e3e0baa0077f1aed55c17f50e7b2397136129fb88f"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.210"
+version = "1.0.215"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "243902eda00fad750862fc144cea25caca5e20d615af0a81bee94ca738f1df1f"
+checksum = "ad1e866f866923f252f05c889987993144fb74e722403468a4ebd70c3cd756c0"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -793,9 +813,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.128"
+version = "1.0.133"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6ff5456707a1de34e7e37f2a6fd3d3f808c318259cbd01ab6377795054b483d8"
+checksum = "c7fceb2473b9166b2294ef05efcb65a3db80803f0b03ef86a5fc88a2b85ee377"
 dependencies = [
  "itoa",
  "memchr",
@@ -907,9 +927,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
 
 [[package]]
 name = "syn"
-version = "2.0.79"
+version = "2.0.87"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89132cd0bf050864e1d38dc3bbc07a0eb8e7530af26344d3d2bbbef83499f590"
+checksum = "25aa4ce346d03a6dcd68dd8b4010bcb74e54e62c90c573f394c46eae99aba32d"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -933,8 +953,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.40.0"
-source = "git+https://github.com/tokio-rs/tokio.git?rev=512e9de#512e9decfb683d22f4a145459142542caa0894c9"
+version = "1.41.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22cfb5bee7a6a52939ca9224d6ac897bb669134078daa8735560897f69de4d33"
 dependencies = [
  "backtrace",
  "bytes",
@@ -1031,9 +1052,9 @@ checksum = "5ab17db44d7388991a428b2ee655ce0c212e862eff1768a455c58f9aad6e7893"
 
 [[package]]
 name = "unicode-ident"
-version = "1.0.13"
+version = "1.0.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e91b56cd4cadaeb79bbf1a5645f6b4f8dc5bde8834ad5894a8db35fda9efa1fe"
+checksum = "adb9e6ca4f869e1180728b7950e35922a7fc6397f7b641499e8f3ef06e50dc83"
 
 [[package]]
 name = "unicode-normalization"
@@ -1248,7 +1269,7 @@ checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
 [[package]]
 name = "xitca-codegen"
 version = "0.4.0"
-source = "git+http://github.com/HFQR/xitca-web?rev=1de8d9c#1de8d9c079e73f7fd9ba953741302d87e50d831a"
+source = "git+http://github.com/HFQR/xitca-web?rev=3b005af#3b005af839ce718f9121d57788f5185296ceacf1"
 dependencies = [
  "quote",
  "syn",
@@ -1257,7 +1278,7 @@ dependencies = [
 [[package]]
 name = "xitca-http"
 version = "0.7.0"
-source = "git+http://github.com/HFQR/xitca-web?rev=1de8d9c#1de8d9c079e73f7fd9ba953741302d87e50d831a"
+source = "git+http://github.com/HFQR/xitca-web?rev=3b005af#3b005af839ce718f9121d57788f5185296ceacf1"
 dependencies = [
  "futures-core",
  "http",
@@ -1307,7 +1328,7 @@ dependencies = [
 [[package]]
 name = "xitca-postgres"
 version = "0.3.0"
-source = "git+http://github.com/HFQR/xitca-web?rev=1de8d9c#1de8d9c079e73f7fd9ba953741302d87e50d831a"
+source = "git+http://github.com/HFQR/xitca-web?rev=3b005af#3b005af839ce718f9121d57788f5185296ceacf1"
 dependencies = [
  "fallible-iterator",
  "futures-core",
@@ -1345,7 +1366,7 @@ dependencies = [
 [[package]]
 name = "xitca-server"
 version = "0.5.0"
-source = "git+http://github.com/HFQR/xitca-web?rev=1de8d9c#1de8d9c079e73f7fd9ba953741302d87e50d831a"
+source = "git+http://github.com/HFQR/xitca-web?rev=3b005af#3b005af839ce718f9121d57788f5185296ceacf1"
 dependencies = [
  "socket2 0.5.7",
  "tokio",
@@ -1359,7 +1380,7 @@ dependencies = [
 [[package]]
 name = "xitca-service"
 version = "0.3.0"
-source = "git+http://github.com/HFQR/xitca-web?rev=1de8d9c#1de8d9c079e73f7fd9ba953741302d87e50d831a"
+source = "git+http://github.com/HFQR/xitca-web?rev=3b005af#3b005af839ce718f9121d57788f5185296ceacf1"
 
 [[package]]
 name = "xitca-unsafe-collection"
@@ -1375,6 +1396,7 @@ name = "xitca-web"
 version = "0.1.0"
 dependencies = [
  "atoi",
+ "core_affinity",
  "diesel",
  "diesel-async",
  "futures-core",
@@ -1400,7 +1422,7 @@ dependencies = [
 [[package]]
 name = "xitca-web"
 version = "0.7.0"
-source = "git+http://github.com/HFQR/xitca-web?rev=1de8d9c#1de8d9c079e73f7fd9ba953741302d87e50d831a"
+source = "git+http://github.com/HFQR/xitca-web?rev=3b005af#3b005af839ce718f9121d57788f5185296ceacf1"
 dependencies = [
  "futures-core",
  "pin-project-lite",

frameworks/Rust/xitca-web/Cargo.toml

@@ -46,7 +46,7 @@ template = ["dep:sailfish"]
 # io-uring optional
 io-uring = ["dep:tokio-uring", "xitca-http/io-uring", "xitca-server/io-uring"]
 # unrealistic performance optimization
-perf = ["dep:mimalloc", "tokio/parking_lot"]
+perf = ["dep:core_affinity", "dep:mimalloc", "tokio/parking_lot"]
 
 [dependencies]
 xitca-http = "0.7"
@@ -81,13 +81,14 @@ sailfish = { version = "0.9", default-features = false, features = ["perf-inline
 tokio-uring = { version = "0.5", optional = true }
 
 # perf optional
+core_affinity = { version = "0.8.1", optional = true }
 mimalloc = { version = "0.1", default-features = false, 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 }
 rand = { version = "0.8", features = ["small_rng"] }
-tokio = "1"
+tokio = "1.41"
 
 [profile.release]
 lto = true
@@ -100,11 +101,10 @@ xitca-postgres-diesel = { git = "https://github.com/fakeshadow/xitca-postgres-di
 
 diesel-async = { git = "https://github.com/weiznich/diesel_async", rev = "5b8262b" }
 mio = { git = "https://github.com/fakeshadow/mio", rev = "9bae6012b7ecfc6083350785f71a5e8265358178" }
-tokio = { git = "https://github.com/tokio-rs/tokio.git", rev = "512e9de" }
-
-xitca-codegen = { git = "http://github.com/HFQR/xitca-web", rev = "1de8d9c" }
-xitca-http = { git = "http://github.com/HFQR/xitca-web", rev = "1de8d9c" }
-xitca-postgres = { git = "http://github.com/HFQR/xitca-web", rev = "1de8d9c" }
-xitca-server = { git = "http://github.com/HFQR/xitca-web", rev = "1de8d9c" }
-xitca-service = { git = "http://github.com/HFQR/xitca-web", rev = "1de8d9c" }
-xitca-web = { git = "http://github.com/HFQR/xitca-web", rev = "1de8d9c" }
+
+xitca-codegen = { git = "http://github.com/HFQR/xitca-web", rev = "3b005af" }
+xitca-http = { git = "http://github.com/HFQR/xitca-web", rev = "3b005af" }
+xitca-postgres = { git = "http://github.com/HFQR/xitca-web", rev = "3b005af" }
+xitca-server = { git = "http://github.com/HFQR/xitca-web", rev = "3b005af" }
+xitca-service = { git = "http://github.com/HFQR/xitca-web", rev = "3b005af" }
+xitca-web = { git = "http://github.com/HFQR/xitca-web", rev = "3b005af" }

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

@@ -51,14 +51,12 @@ where
 #[cold]
 #[inline(never)]
 fn error_handler(e: RouterError<util::Error>) -> Response {
-    match e {
-        RouterError::Match(_) => error_response(StatusCode::NOT_FOUND),
-        RouterError::NotAllowed(_) => error_response(StatusCode::METHOD_NOT_ALLOWED),
-        RouterError::Service(e) => {
-            println!("{e}");
-            error_response(StatusCode::INTERNAL_SERVER_ERROR)
-        }
-    }
+    let status = match e {
+        RouterError::Match(_) => StatusCode::NOT_FOUND,
+        RouterError::NotAllowed(_) => StatusCode::METHOD_NOT_ALLOWED,
+        RouterError::Service(_) => StatusCode::INTERNAL_SERVER_ERROR,
+    };
+    error_response(status)
 }
 
 async fn plain_text(ctx: Ctx<'_>) -> HandleResult<Response> {

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

@@ -10,14 +10,13 @@ mod db;
 mod ser;
 mod util;
 
-use std::{convert::Infallible, io};
+use std::io;
 
 use xitca_http::{
     bytes::BufMutWriter,
     h1::dispatcher_unreal::{Dispatcher, Request, Response},
     http::StatusCode,
 };
-use xitca_io::net::TcpStream;
 use xitca_service::Service;
 
 use self::{
@@ -30,49 +29,57 @@ fn main() -> io::Result<()> {
 
     let cores = std::thread::available_parallelism().map(|num| num.get()).unwrap_or(56);
 
+    let mut ids = core_affinity::get_core_ids().unwrap();
+
+    let worker = move |id: Option<core_affinity::CoreId>| {
+        if let Some(id) = id {
+            let _ = core_affinity::set_for_current(id);
+        }
+
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build_local(&Default::default())
+            .unwrap()
+            .block_on(async {
+                let socket = tokio::net::TcpSocket::new_v4()?;
+                socket.set_reuseaddr(true)?;
+                // unrealistic due to following reason:
+                // 1. this only works good on unix system.
+                // 2. no resource distribution adjustment between sockets on different threads. causing uneven workload
+                // where some threads are idle while others busy. resulting in overall increased latency
+                socket.set_reuseport(true)?;
+                socket.bind(addr)?;
+                let listener = socket.listen(1024)?;
+
+                let client = db::create().await.unwrap();
+
+                // unrealistic http dispatcher. no spec check. no security feature.
+                let service = Dispatcher::new(handler, State::new(client));
+
+                loop {
+                    match listener.accept().await {
+                        Ok((stream, _)) => {
+                            let service = service.clone();
+                            tokio::task::spawn_local(async move {
+                                let _ = service.call(stream.into()).await;
+                            });
+                        }
+                        Err(e) => return Err(e),
+                    };
+                }
+            })
+    };
+
     let handle = core::iter::repeat_with(|| {
-        std::thread::spawn(move || {
-            tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build_local(&Default::default())
-                .unwrap()
-                .block_on(async {
-                    let socket = tokio::net::TcpSocket::new_v4()?;
-                    socket.set_reuseaddr(true)?;
-                    // unrealistic due to following reason:
-                    // 1. this only works good on unix system.
-                    // 2. no resource distribution adjustment between sockets on different threads. causing uneven workload
-                    // where some threads are idle while others busy. resulting in overall increased latency
-                    socket.set_reuseport(true)?;
-                    socket.bind(addr)?;
-                    let listener = socket.listen(1024)?;
-
-                    let client = db::create().await.unwrap();
-
-                    // unrealistic http dispatcher. no spec check. no security feature.
-                    let service = Dispatcher::new(handler, State::new(client));
-
-                    loop {
-                        match listener.accept().await {
-                            Ok((stream, _)) => {
-                                let stream = stream.into_std()?;
-                                let stream = TcpStream::from_std(stream)?;
-                                let service = service.clone();
-                                tokio::task::spawn_local(async move {
-                                    let _ = service.call(stream).await;
-                                });
-                            }
-                            Err(e) => return Err(e),
-                        };
-                    }
-                })
-        })
+        let id = ids.pop();
+        std::thread::spawn(move || worker(id))
     })
-    .take(cores)
+    .take(cores - 1)
     .collect::<Vec<_>>();
 
     // unrealistic due to no signal handling, not shutdown handling. when killing this process all resources that
     // need clean async shutdown will be leaked.
+    worker(ids.pop())?;
     for handle in handle {
         handle.join().unwrap()?;
     }
@@ -80,7 +87,7 @@ fn main() -> io::Result<()> {
     Ok(())
 }
 
-async fn handler<'h>(req: Request<'h>, res: Response<'h>, state: &State<db::Client>) -> Response<'h, 3> {
+async fn handler<'h>(req: Request<'h, State<db::Client>>, res: Response<'h>) -> Response<'h, 3> {
     // unrealistic due to no http method check
     match req.path {
         // unrealistic due to no dynamic path matching
@@ -91,8 +98,7 @@ async fn handler<'h>(req: Request<'h>, res: Response<'h>, state: &State<db::Clie
                 .header("server", "X")
                 // unrealistic content length header.
                 .header("content-length", "13")
-                .body_writer(|buf| Ok::<_, Infallible>(buf.extend_from_slice(b"Hello, World!")))
-                .unwrap()
+                .body_writer(|buf| buf.extend_from_slice(b"Hello, World!"))
         }
         "/json" => res
             .status(StatusCode::OK)
@@ -100,12 +106,12 @@ async fn handler<'h>(req: Request<'h>, res: Response<'h>, state: &State<db::Clie
             .header("server", "X")
             // unrealistic content length header.
             .header("content-length", "27")
-            .body_writer(|buf| serde_json::to_writer(BufMutWriter(buf), &Message::new()))
-            .unwrap(),
+            .body_writer(|buf| serde_json::to_writer(BufMutWriter(buf), &Message::new()).unwrap()),
+
         // all database related categories are unrealistic. please reference db_unrealistic module for detail.
         "/fortunes" => {
             use sailfish::TemplateOnce;
-            let fortunes = state.client.tell_fortune().await.unwrap().render_once().unwrap();
+            let fortunes = req.ctx.client.tell_fortune().await.unwrap().render_once().unwrap();
             res.status(StatusCode::OK)
                 .header("content-type", "text/html; charset=utf-8")
                 .header("server", "X")
@@ -114,18 +120,18 @@ async fn handler<'h>(req: Request<'h>, res: Response<'h>, state: &State<db::Clie
         "/db" => {
             // unrealistic due to no error handling. any db/serialization error will cause process crash.
             // the same goes for all following unwraps on database related functions.
-            let world = state.client.get_world().await.unwrap();
-            json_response(res, state, &world)
+            let world = req.ctx.client.get_world().await.unwrap();
+            json_response(res, req.ctx, &world)
         }
         p if p.starts_with("/q") => {
             let num = p["/queries?q=".len()..].parse_query();
-            let worlds = state.client.get_worlds(num).await.unwrap();
-            json_response(res, state, &worlds)
+            let worlds = req.ctx.client.get_worlds(num).await.unwrap();
+            json_response(res, req.ctx, &worlds)
         }
         p if p.starts_with("/u") => {
             let num = p["/updates?q=".len()..].parse_query();
-            let worlds = state.client.update(num).await.unwrap();
-            json_response(res, state, &worlds)
+            let worlds = req.ctx.client.update(num).await.unwrap();
+            json_response(res, req.ctx, &worlds)
         }
         _ => res.status(StatusCode::NOT_FOUND).header("server", "X").body(&[]),
     }