From d659117ae51caa399a6e44d5024305db41f87fef Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= <joakim.frostegard@gmail.com>
Date: Wed, 27 Oct 2021 20:19:02 +0200
Subject: [PATCH] Move CoreAffinityConfig to aquatic_common, use in glommio
 http

---
 Cargo.lock                          |  1 +
 aquatic_common/src/cpu_pinning.rs   | 17 +++++++++++++++++
 aquatic_common/src/lib.rs           |  1 +
 aquatic_http/Cargo.toml             |  1 +
 aquatic_http/src/lib/config.rs      |  3 +++
 aquatic_http/src/lib/glommio/mod.rs | 20 +++++++++++++-------
 aquatic_udp/src/lib/config.rs       | 17 +----------------
 7 files changed, 37 insertions(+), 23 deletions(-)
 create mode 100644 aquatic_common/src/cpu_pinning.rs

diff --git a/Cargo.lock b/Cargo.lock
index d4959b1..b8de54a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -92,6 +92,7 @@ dependencies = [
  "aquatic_common",
  "aquatic_http_protocol",
  "cfg-if",
+ "core_affinity",
  "crossbeam-channel",
  "either",
  "futures-lite",
diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs
new file mode 100644
index 0000000..e76f652
--- /dev/null
+++ b/aquatic_common/src/cpu_pinning.rs
@@ -0,0 +1,17 @@
+use serde::{Serialize, Deserialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(default)]
+pub struct CoreAffinityConfig {
+    pub set_affinities: bool,
+    pub offset: usize,
+}
+
+impl Default for CoreAffinityConfig {
+    fn default() -> Self {
+        Self {
+            set_affinities: false,
+            offset: 0,
+        }
+    }
+}
diff --git a/aquatic_common/src/lib.rs b/aquatic_common/src/lib.rs
index 3e90bbc..4ad9e8e 100644
--- a/aquatic_common/src/lib.rs
+++ b/aquatic_common/src/lib.rs
@@ -5,6 +5,7 @@ use indexmap::IndexMap;
 use rand::Rng;
 
 pub mod access_list;
+pub mod cpu_pinning;
 
 /// Peer or connection valid until this instant
 ///
diff --git a/aquatic_http/Cargo.toml b/aquatic_http/Cargo.toml
index 97f7bab..49cac1c 100644
--- a/aquatic_http/Cargo.toml
+++ b/aquatic_http/Cargo.toml
@@ -26,6 +26,7 @@ aquatic_cli_helpers = "0.1.0"
 aquatic_common = "0.1.0"
 aquatic_http_protocol = "0.1.0"
 cfg-if = "1"
+core_affinity = "0.5"
 either = "1"
 hashbrown = "0.11.2"
 indexmap = "1"
diff --git a/aquatic_http/src/lib/config.rs b/aquatic_http/src/lib/config.rs
index 08ccfad..e725dbf 100644
--- a/aquatic_http/src/lib/config.rs
+++ b/aquatic_http/src/lib/config.rs
@@ -1,6 +1,7 @@
 use std::{net::SocketAddr, path::PathBuf};
 
 use aquatic_common::access_list::AccessListConfig;
+use aquatic_common::cpu_pinning::CoreAffinityConfig;
 use serde::{Deserialize, Serialize};
 
 use aquatic_cli_helpers::LogLevel;
@@ -23,6 +24,7 @@ pub struct Config {
     pub statistics: StatisticsConfig,
     pub privileges: PrivilegeConfig,
     pub access_list: AccessListConfig,
+    pub core_affinity: CoreAffinityConfig,
 }
 
 impl aquatic_cli_helpers::Config for Config {
@@ -116,6 +118,7 @@ impl Default for Config {
             statistics: StatisticsConfig::default(),
             privileges: PrivilegeConfig::default(),
             access_list: AccessListConfig::default(),
+            core_affinity: CoreAffinityConfig::default(),
         }
     }
 }
diff --git a/aquatic_http/src/lib/glommio/mod.rs b/aquatic_http/src/lib/glommio/mod.rs
index c93fbc5..f53ed78 100644
--- a/aquatic_http/src/lib/glommio/mod.rs
+++ b/aquatic_http/src/lib/glommio/mod.rs
@@ -16,6 +16,12 @@ mod network;
 const SHARED_CHANNEL_SIZE: usize = 1024;
 
 pub fn run(config: Config) -> anyhow::Result<()> {
+    if config.core_affinity.set_affinities {
+        core_affinity::set_for_current(core_affinity::CoreId {
+            id: config.core_affinity.offset,
+        });
+    }
+
     let access_list = if config.access_list.mode.is_on() {
         AccessList::create_from_path(&config.access_list.path).expect("Load access list")
     } else {
@@ -43,9 +49,9 @@ pub fn run(config: Config) -> anyhow::Result<()> {
 
         let mut builder = LocalExecutorBuilder::default();
 
-        // if config.core_affinity.set_affinities {
-        //     builder = builder.pin_to_cpu(config.core_affinity.offset + 1 + i);
-        // }
+        if config.core_affinity.set_affinities {
+            builder = builder.pin_to_cpu(config.core_affinity.offset + 1 + i);
+        }
 
         let executor = builder.spawn(|| async move {
             network::run_socket_worker(
@@ -70,10 +76,10 @@ pub fn run(config: Config) -> anyhow::Result<()> {
 
         let mut builder = LocalExecutorBuilder::default();
 
-        // if config.core_affinity.set_affinities {
-        //     builder =
-        //         builder.pin_to_cpu(config.core_affinity.offset + 1 + config.socket_workers + i);
-        // }
+        if config.core_affinity.set_affinities {
+            builder =
+                builder.pin_to_cpu(config.core_affinity.offset + 1 + config.socket_workers + i);
+        }
 
         let executor = builder.spawn(|| async move {
             handlers::run_request_worker(
diff --git a/aquatic_udp/src/lib/config.rs b/aquatic_udp/src/lib/config.rs
index d05b338..f897a96 100644
--- a/aquatic_udp/src/lib/config.rs
+++ b/aquatic_udp/src/lib/config.rs
@@ -1,6 +1,7 @@
 use std::net::SocketAddr;
 
 use aquatic_common::access_list::AccessListConfig;
+use aquatic_common::cpu_pinning::CoreAffinityConfig;
 use serde::{Deserialize, Serialize};
 
 use aquatic_cli_helpers::LogLevel;
@@ -109,13 +110,6 @@ pub struct PrivilegeConfig {
     pub user: String,
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize)]
-#[serde(default)]
-pub struct CoreAffinityConfig {
-    pub set_affinities: bool,
-    pub offset: usize,
-}
-
 impl Default for Config {
     fn default() -> Self {
         Self {
@@ -193,12 +187,3 @@ impl Default for PrivilegeConfig {
         }
     }
 }
-
-impl Default for CoreAffinityConfig {
-    fn default() -> Self {
-        Self {
-            set_affinities: false,
-            offset: 0,
-        }
-    }
-}