cpu affinity

Author: fafhrd91

frameworks/Rust/ntex/Cargo.toml

@@ -37,8 +37,8 @@ tokio = ["ntex/tokio"]
 compio = ["ntex/compio", ]
 
 [dependencies]
-ntex = "2.4"
-ntex-compio = "0.1.2"
+ntex = "2.8"
+ntex-compio = "0.2"
 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"] }
@@ -47,16 +47,19 @@ buf-min = { version = "0.7", features = ["ntex-bytes"] }
 env_logger = "0.11"
 nanorand = { version = "0.7", default-features = false, features = ["std", "wyrand", "tls"] }
 atoi = "2.0"
-num_cpus = "1.16"
+core_affinity = "0.8"
 smallvec = "1.13"
 futures = "0.3"
+sonic-rs = "0.3.16"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 log = { version = "0.4", features = ["release_max_level_off"] }
-compio-driver = { version = "0.4", features = ["io-uring", "io-uring-socket"]}
 tok_io = {version = "1", package = "tokio" }
 tokio-postgres = { git="https://github.com/fafhrd91/postgres.git", branch="ntex-2" }
 
+[target.'cfg(target_os = "linux")'.dependencies]
+compio-driver = { version = "*", features = ["io-uring"]}
+
 [profile.release]
 opt-level = 3
 codegen-units = 1

frameworks/Rust/ntex/src/main.rs

@@ -2,7 +2,7 @@
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
 use ntex::http::header::{CONTENT_TYPE, SERVER};
-use ntex::{http, time::Seconds, util::BytesMut, util::PoolId, web};
+use ntex::{http, time::Seconds, util::BytesMut, util::PoolId, util::Ready, web};
 use yarte::Serialize;
 
 mod utils;
@@ -45,6 +45,10 @@ async fn plaintext() -> web::HttpResponse {
 async fn main() -> std::io::Result<()> {
     println!("Started http server: 127.0.0.1:8080");
 
+    let cores = core_affinity::get_core_ids().unwrap();
+    let total_cores = cores.len();
+    let cores = std::sync::Arc::new(std::sync::Mutex::new(cores));
+
     // start http server
     ntex::server::build()
         .backlog(1024)
@@ -60,7 +64,17 @@ async fn main() -> std::io::Result<()> {
                 .payload_read_rate(Seconds::ZERO, Seconds::ZERO, 0)
                 .h1(web::App::new().service(json).service(plaintext).finish())
         })?
-        .workers(num_cpus::get())
+        .configure(move |cfg| {
+            let cores = cores.clone();
+            cfg.on_worker_start(move |_| {
+                if let Some(core) = cores.lock().unwrap().pop() {
+                    core_affinity::set_for_current(core);
+                }
+                Ready::<_, &str>::Ok(())
+            });
+            Ok(())
+        })?
+        .workers(total_cores)
         .run()
         .await
 }

frameworks/Rust/ntex/src/main_db.rs

@@ -1,13 +1,12 @@
 #[cfg(not(target_os = "macos"))]
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
-// static GLOBAL: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
 
 use ntex::http::header::{CONTENT_TYPE, SERVER};
 use ntex::http::{HttpService, KeepAlive, Request, Response, StatusCode};
 use ntex::service::{Service, ServiceCtx, ServiceFactory};
 use ntex::web::{Error, HttpResponse};
-use ntex::{time::Seconds, util::PoolId};
+use ntex::{time::Seconds, util::PoolId, util::Ready};
 
 mod db;
 mod utils;
@@ -83,6 +82,10 @@ impl ServiceFactory<Request> for AppFactory {
 async fn main() -> std::io::Result<()> {
     println!("Starting http server: 127.0.0.1:8080");
 
+    let cores = core_affinity::get_core_ids().unwrap();
+    let total_cores = cores.len();
+    let cores = std::sync::Arc::new(std::sync::Mutex::new(cores));
+
     ntex::server::build()
         .backlog(1024)
         .bind("techempower", "0.0.0.0:8080", |cfg| {
@@ -97,7 +100,17 @@ async fn main() -> std::io::Result<()> {
                 .payload_read_rate(Seconds::ZERO, Seconds::ZERO, 0)
                 .h1(AppFactory)
         })?
-        .workers(num_cpus::get())
+        .configure(move |cfg| {
+            let cores = cores.clone();
+            cfg.on_worker_start(move |_| {
+                if let Some(core) = cores.lock().unwrap().pop() {
+                    core_affinity::set_for_current(core);
+                }
+                Ready::<_, &str>::Ok(())
+            });
+            Ok(())
+        })?
+        .workers(total_cores)
         .run()
         .await
 }

frameworks/Rust/ntex/src/main_plt.rs

@@ -3,7 +3,8 @@ static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
 use std::{future::Future, io, pin::Pin, task::Context, task::Poll};
 
-use ntex::{fn_service, http::h1, io::Io, io::RecvError, util::ready, util::PoolId};
+use ntex::util::{ready, PoolId, Ready};
+use ntex::{fn_service, http::h1, io::Io, io::RecvError};
 use yarte::Serialize;
 
 mod utils;
@@ -75,6 +76,10 @@ impl Future for App {
 async fn main() -> io::Result<()> {
     println!("Started http server: 127.0.0.1:8080");
 
+    let cores = core_affinity::get_core_ids().unwrap();
+    let total_cores = cores.len();
+    let cores = std::sync::Arc::new(std::sync::Mutex::new(cores));
+
     // start http server
     ntex::server::build()
         .backlog(1024)
@@ -88,7 +93,17 @@ async fn main() -> io::Result<()> {
                 codec: h1::Codec::default(),
             })
         })?
-        .workers(num_cpus::get())
+        .configure(move |cfg| {
+            let cores = cores.clone();
+            cfg.on_worker_start(move |_| {
+                if let Some(core) = cores.lock().unwrap().pop() {
+                    core_affinity::set_for_current(core);
+                }
+                Ready::<_, &str>::Ok(())
+            });
+            Ok(())
+        })?
+        .workers(total_cores)
         .run()
         .await
 }