perf(engine): set thread priority to max with realtime RR policy

Engine thread is the one doing EVM execution, and it should have the
highest priority out of all other threads (RPC / prewarming / state root task).

This adds a new spawn_os_thread_with_priority function to reth_tasks that
spawns threads with realtime round-robin scheduling and max priority.
The engine task now uses this to ensure optimal EVM execution performance.

Amp-Thread-ID: https://ampcode.com/threads/T-019c2aea-dd57-718e-97cd-ac206cf67cd1
Co-authored-by: Amp <amp@ampcode.com>

Cargo.toml

@@ -583,6 +583,7 @@ syn = "2.0"
 thiserror = { version = "2.0.0", default-features = false }
 tar = "0.4.44"
 tracing = { version = "0.1.0", default-features = false }
+thread-priority = "1"
 tracing-appender = "0.2"
 url = { version = "2.3", default-features = false }
 zstd = "0.13"

crates/engine/tree/src/tree/mod.rs

@@ -37,7 +37,7 @@ use reth_provider::{
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_stages_api::ControlFlow;
-use reth_tasks::spawn_os_thread;
+use reth_tasks::spawn_os_thread_with_priority;
 use reth_trie_db::ChangesetCache;
 use revm::state::EvmState;
 use state::TreeState;
@@ -432,7 +432,7 @@ where
             changeset_cache,
         );
         let incoming = task.incoming_tx.clone();
-        spawn_os_thread("engine", || task.run());
+        spawn_os_thread_with_priority("engine", || task.run());
         (incoming, outgoing)
     }
 

crates/tasks/Cargo.toml

@@ -26,6 +26,7 @@ auto_impl.workspace = true
 tracing.workspace = true
 thiserror.workspace = true
 dyn-clone.workspace = true
+thread-priority.workspace = true
 
 # feature `rayon`
 rayon = { workspace = true, optional = true }

crates/tasks/src/lib.rs

@@ -32,12 +32,15 @@ use std::{
     task::{ready, Context, Poll},
     thread,
 };
+use thread_priority::{
+    RealtimeThreadSchedulePolicy, ThreadBuilder, ThreadPriority, ThreadSchedulePolicy,
+};
 use tokio::{
     runtime::Handle,
     sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
     task::JoinHandle,
 };
-use tracing::{debug, error};
+use tracing::{debug, error, warn};
 use tracing_futures::Instrument;
 
 pub mod metrics;
@@ -94,6 +97,35 @@ where
         .unwrap_or_else(|e| panic!("failed to spawn scoped thread {name:?}: {e}"))
 }
 
+/// Spawns an OS thread with realtime round-robin scheduling policy and max priority.
+///
+/// This function is similar to [`spawn_os_thread`] but additionally sets the thread to use
+/// realtime round-robin scheduling with maximum priority. This is useful for latency-critical
+/// tasks like EVM execution where the thread should have priority over other tasks.
+///
+/// If setting the thread priority fails, a warning is logged but the thread continues to run.
+#[track_caller]
+pub fn spawn_os_thread_with_priority<F, T>(name: &str, f: F) -> thread::JoinHandle<T>
+where
+    F: FnOnce() -> T + Send + 'static,
+    T: Send + 'static,
+{
+    let handle = Handle::try_current().ok();
+    let thread_name = name.to_string();
+    ThreadBuilder::default()
+        .name(&thread_name)
+        .policy(ThreadSchedulePolicy::Realtime(RealtimeThreadSchedulePolicy::RoundRobin))
+        .priority(ThreadPriority::Max)
+        .spawn(move |result| {
+            if let Err(err) = result {
+                warn!(target: "reth::tasks", ?err, "Failed to set thread priority for {}", thread_name);
+            }
+            let _guard = handle.as_ref().map(Handle::enter);
+            f()
+        })
+        .unwrap_or_else(|e| panic!("failed to spawn thread {name:?}: {e}"))
+}
+
 /// A type that can spawn tasks.
 ///
 /// The main purpose of this type is to abstract over [`TaskExecutor`] so it's more convenient to