reth_tasks/
utils.rs

1//! Task utility functions.
2
3pub use thread_priority::{self, *};
4
5/// Runs the given closure exactly once per call site.
6///
7/// Each invocation expands to its own `static Once`, so two `once!` calls in the same function
8/// are independent. Useful for one-time thread setup (priority, affinity) on threads that may
9/// be re-entered (e.g. tokio blocking pool).
10#[macro_export]
11macro_rules! once {
12    ($e:expr) => {{
13        static ONCE: std::sync::Once = std::sync::Once::new();
14        ONCE.call_once($e);
15    }};
16}
17
18/// Increases the current thread's priority.
19///
20/// Tries [`ThreadPriority::Max`] first. If that fails (e.g. missing `CAP_SYS_NICE`),
21/// falls back to a moderate bump via [`ThreadPriority::Crossplatform`] (~5 nice points
22/// on unix). Failures are logged at `debug` level.
23pub fn increase_thread_priority() {
24    let thread_name = std::thread::current().name().unwrap_or("unnamed").to_string();
25    if let Err(err) = ThreadPriority::Max.set_for_current() {
26        tracing::debug!(%thread_name, ?err, "failed to set max thread priority, trying moderate bump; grant CAP_SYS_NICE to the process to enable this");
27        // Crossplatform value 62/99 ≈ nice -5 on unix.
28        let fallback = ThreadPriority::Crossplatform(
29            ThreadPriorityValue::try_from(62u8).expect("62 is within the valid 0..100 range"),
30        );
31        if let Err(err) = fallback.set_for_current() {
32            tracing::debug!(%thread_name, ?err, "failed to set moderate thread priority");
33        }
34    }
35}
36
37/// Deprioritizes known background threads spawned by third-party libraries (`tracing-appender`,
38/// `reqwest`) by scanning `/proc/<pid>/task/` for matching thread names and setting `SCHED_IDLE`
39/// scheduling policy on them.
40///
41/// This is a hack: these threads are spawned by libraries that do not expose a way to hook into
42/// thread initialization or expose the TIDs, so we have to discover them after the fact by
43/// reading `/proc`.
44///
45/// Should be called once after tracing is initialized.
46///
47/// No-op on non-Linux platforms.
48#[allow(clippy::missing_const_for_fn)]
49pub fn deprioritize_background_threads() {
50    #[cfg(target_os = "linux")]
51    _deprioritize_background_threads();
52}
53
54/// Thread name prefixes to deprioritize.
55///
56/// Do not re-add the OTLP span exporter (`OpenTelemetry.T`) to avoid CPU spinning.
57/// See <https://github.com/paradigmxyz/reth/pull/27370>.
58#[cfg(target_os = "linux")]
59const DEPRIORITIZE_THREAD_PREFIXES: &[&str] = &["tracing-appende", "reqwest-interna"];
60
61#[cfg(target_os = "linux")]
62fn _deprioritize_background_threads() {
63    let pid = std::process::id();
64    let task_dir = format!("/proc/{pid}/task");
65
66    let entries = match std::fs::read_dir(&task_dir) {
67        Ok(entries) => entries,
68        Err(err) => {
69            tracing::debug!(%err, "failed to read /proc task directory");
70            return;
71        }
72    };
73
74    for entry in entries.filter_map(Result::ok) {
75        let tid_str = entry.file_name();
76        let Some(tid_str) = tid_str.to_str() else { continue };
77        let Ok(tid) = tid_str.parse::<i32>() else { continue };
78
79        let comm_path = format!("{task_dir}/{tid_str}/comm");
80        let comm = match std::fs::read_to_string(&comm_path) {
81            Ok(c) => c,
82            Err(_) => continue,
83        };
84        let comm = comm.trim();
85
86        if !DEPRIORITIZE_THREAD_PREFIXES.iter().any(|prefix| comm.starts_with(prefix)) {
87            continue;
88        }
89
90        // SCHED_IDLE is the lowest-priority scheduling class. The kernel will only schedule these
91        // threads when no other (SCHED_OTHER/SCHED_BATCH/RT) threads need the CPU.
92        // SAFETY: sched_setscheduler is safe to call with a valid TID.
93        unsafe {
94            let param = libc::sched_param { sched_priority: 0 };
95            if libc::sched_setscheduler(tid, libc::SCHED_IDLE, std::ptr::from_ref(&param)) != 0 {
96                tracing::debug!(
97                    tid,
98                    comm,
99                    err = std::io::Error::last_os_error().to_string(),
100                    "failed to set SCHED_IDLE"
101                );
102            }
103        }
104
105        tracing::debug!(tid, comm, "deprioritized background thread (SCHED_IDLE)");
106    }
107}
reth_tasks/utils.rs

reth_tasks/
utils.rs
