Merge pull request #2566 from subspace/thread-priority

Thread priority
autonomys · Feb 29, 2024 · 4b690d9 · 4b690d9
2 parents 323b199 + 4c37052
commit 4b690d9
Show file tree

Hide file tree

Showing 6 changed files with 91 additions and 3 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/sc-proof-of-time/Cargo.toml b/crates/sc-proof-of-time/Cargo.toml
@@ -17,6 +17,8 @@ derive_more = "0.99.17"
 futures = "0.3.29"
 lru = "0.12.1"
 parity-scale-codec = { version = "3.6.9", features = ["derive"] }
+parking_lot = "0.12.1"
+rayon = "1.8.1"
 sc-client-api = { version = "4.0.0-dev", git = "https://github.com/subspace/polkadot-sdk", rev = "d6b500960579d73c43fc4ef550b703acfa61c4c8" }
 sc-consensus-slots = { version = "0.10.0-dev", git = "https://github.com/subspace/polkadot-sdk", rev = "d6b500960579d73c43fc4ef550b703acfa61c4c8" }
 sc-network = { version = "0.10.0-dev", git = "https://github.com/subspace/polkadot-sdk", rev = "d6b500960579d73c43fc4ef550b703acfa61c4c8" }
@@ -30,6 +32,5 @@ sp-inherents = { version = "4.0.0-dev", git = "https://github.com/subspace/polka
 sp-runtime = { version = "24.0.0", git = "https://github.com/subspace/polkadot-sdk", rev = "d6b500960579d73c43fc4ef550b703acfa61c4c8" }
 subspace-core-primitives = { version = "0.1.0", path = "../subspace-core-primitives" }
 subspace-proof-of-time = { version = "0.1.0", path = "../subspace-proof-of-time" }
-parking_lot = "0.12.1"
-rayon = "1.8.1"
+thread-priority = "0.16.0"
 tracing = "0.1.40"
diff --git a/crates/sc-proof-of-time/src/source.rs b/crates/sc-proof-of-time/src/source.rs
@@ -28,6 +28,7 @@ use std::marker::PhantomData;
 use std::sync::Arc;
 use std::thread;
 use subspace_core_primitives::PotCheckpoints;
+use thread_priority::{set_current_thread_priority, ThreadPriority};
 use tracing::{debug, error, trace, warn};
 
 const LOCAL_PROOFS_CHANNEL_CAPACITY: usize = 10;
@@ -150,6 +151,14 @@ where
                         }
                     }
 
+                    if let Err(error) = set_current_thread_priority(ThreadPriority::Max) {
+                        warn!(
+                            %error,
+                            "Failed to set thread priority, timekeeper performance may be \
+                            negatively impacted by other software running on this machine",
+                        );
+                    }
+
                     if let Err(error) =
                         run_timekeeper(state, pot_verifier, timekeeper_proofs_sender)
                     {

diff --git a/crates/subspace-farmer/Cargo.toml b/crates/subspace-farmer/Cargo.toml
@@ -57,6 +57,7 @@ substrate-bip39 = "0.4.5"
 supports-color = "2.1.0"
 tempfile = "3.9.0"
 thiserror = "1.0.56"
+thread-priority = "0.16.0"
 tokio = { version = "1.35.1", features = ["macros", "parking_lot", "rt-multi-thread", "signal", "time"] }
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }

diff --git a/crates/subspace-farmer/src/bin/subspace-farmer/commands/farm.rs b/crates/subspace-farmer/src/bin/subspace-farmer/commands/farm.rs
@@ -13,14 +13,14 @@ use futures::stream::{FuturesOrdered, FuturesUnordered};
 use futures::{FutureExt, StreamExt};
 use parking_lot::Mutex;
 use prometheus_client::registry::Registry;
-use std::fs;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::num::{NonZeroU8, NonZeroUsize};
 use std::path::PathBuf;
 use std::pin::pin;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
+use std::{fmt, fs};
 use subspace_core_primitives::crypto::kzg::{embedded_kzg_settings, Kzg};
 use subspace_core_primitives::{PublicKey, Record, SectorIndex};
 use subspace_erasure_coding::ErasureCoding;
@@ -47,6 +47,7 @@ use subspace_networking::libp2p::multiaddr::Protocol;
 use subspace_networking::libp2p::Multiaddr;
 use subspace_networking::utils::piece_provider::PieceProvider;
 use subspace_proof_of_space::Table;
+use thread_priority::ThreadPriority;
 use tokio::sync::Semaphore;
 use tracing::{debug, error, info, info_span, warn};
 use zeroize::Zeroizing;
@@ -66,6 +67,50 @@ fn should_farm_during_initial_plotting() -> bool {
     total_cpu_cores > 8
 }
 
+/// Plotting thread priority
+#[derive(Debug, Parser, Copy, Clone)]
+enum PlottingThreadPriority {
+    /// Minimum priority
+    Min,
+    /// Default priority
+    Default,
+    /// Max priority (not recommended)
+    Max,
+}
+
+impl FromStr for PlottingThreadPriority {
+    type Err = String;
+
+    fn from_str(s: &str) -> anyhow::Result<Self, Self::Err> {
+        match s {
+            "min" => Ok(Self::Min),
+            "default" => Ok(Self::Default),
+            "max" => Ok(Self::Max),
+            s => Err(format!("Thread priority {s} is not valid")),
+        }
+    }
+}
+
+impl fmt::Display for PlottingThreadPriority {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::Min => "min",
+            Self::Default => "default",
+            Self::Max => "max",
+        })
+    }
+}
+
+impl From<PlottingThreadPriority> for Option<ThreadPriority> {
+    fn from(value: PlottingThreadPriority) -> Self {
+        match value {
+            PlottingThreadPriority::Min => Some(ThreadPriority::Min),
+            PlottingThreadPriority::Default => None,
+            PlottingThreadPriority::Max => Some(ThreadPriority::Max),
+        }
+    }
+}
+
 /// Arguments for farmer
 #[derive(Debug, Parser)]
 pub(crate) struct FarmingArgs {
@@ -193,6 +238,10 @@ pub(crate) struct FarmingArgs {
     /// each with a pair of CPU cores.
     #[arg(long, conflicts_with_all = & ["sector_encoding_concurrency", "replotting_thread_pool_size"])]
     replotting_cpu_cores: Option<String>,
+    /// Plotting thread priority, by default de-prioritizes plotting threads in order to make sure
+    /// farming is successful and computer can be used comfortably for other things
+    #[arg(long, default_value_t = PlottingThreadPriority::Min)]
+    plotting_thread_priority: PlottingThreadPriority,
     /// Disable farm locking, for example if file system doesn't support it
     #[arg(long)]
     disable_farm_locking: bool,
@@ -346,6 +395,7 @@ where
         plotting_cpu_cores,
         replotting_thread_pool_size,
         replotting_cpu_cores,
+        plotting_thread_priority,
         disable_farm_locking,
     } = farming_args;
 
@@ -571,6 +621,7 @@ where
         plotting_thread_pool_core_indices
             .into_iter()
             .zip(replotting_thread_pool_core_indices),
+        plotting_thread_priority.into(),
     )?;
     let farming_thread_pool_size = farming_thread_pool_size
         .map(|farming_thread_pool_size| farming_thread_pool_size.get())

diff --git a/crates/subspace-farmer/src/utils.rs b/crates/subspace-farmer/src/utils.rs
@@ -17,6 +17,7 @@ use std::pin::{pin, Pin};
 use std::str::FromStr;
 use std::task::{Context, Poll};
 use std::{io, thread};
+use thread_priority::{set_current_thread_priority, ThreadPriority};
 use tokio::runtime::Handle;
 use tokio::task;
 use tracing::debug;
@@ -371,6 +372,7 @@ fn create_plotting_thread_pool_manager_thread_pool_pair(
     thread_prefix: &'static str,
     thread_pool_index: usize,
     cpu_core_set: CpuCoreSet,
+    thread_priority: Option<ThreadPriority>,
 ) -> Result<ThreadPool, ThreadPoolBuildError> {
     ThreadPoolBuilder::new()
         .thread_name(move |thread_index| {
@@ -386,6 +388,11 @@ fn create_plotting_thread_pool_manager_thread_pool_pair(
 
                 move || {
                     cpu_core_set.pin_current_thread();
+                    if let Some(thread_priority) = thread_priority {
+                        if let Err(error) = set_current_thread_priority(thread_priority) {
+                            warn!(%error, "Failed to set thread priority");
+                        }
+                    }
                     drop(cpu_core_set);
 
                     let _guard = handle.enter();
@@ -405,6 +412,7 @@ fn create_plotting_thread_pool_manager_thread_pool_pair(
 /// support for user customizations is desired. They will then have to be composed into pairs for this function.
 pub fn create_plotting_thread_pool_manager<I>(
     mut cpu_core_sets: I,
+    thread_priority: Option<ThreadPriority>,
 ) -> Result<PlottingThreadPoolManager, ThreadPoolBuildError>
 where
     I: ExactSizeIterator<Item = (CpuCoreSet, CpuCoreSet)>,
@@ -422,11 +430,13 @@ where
                     "plotting",
                     thread_pool_index,
                     plotting_cpu_core_set,
+                    thread_priority,
                 )?,
                 replotting: create_plotting_thread_pool_manager_thread_pool_pair(
                     "replotting",
                     thread_pool_index,
                     replotting_cpu_core_set,
+                    thread_priority,
                 )?,
             })
         },