Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 2 additions & 17 deletions crates/cli/runner/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,15 +11,7 @@
//! Entrypoint for running commands.

use reth_tasks::{TaskExecutor, TaskManager};
use std::{
future::Future,
pin::pin,
sync::{
atomic::{AtomicUsize, Ordering},
mpsc,
},
time::Duration,
};
use std::{future::Future, pin::pin, sync::mpsc, time::Duration};
use tracing::{debug, error, trace};

/// Executes CLI commands.
Expand Down Expand Up @@ -167,14 +159,7 @@ pub struct CliContext {
/// Creates a new default tokio multi-thread [Runtime](tokio::runtime::Runtime) with all features
/// enabled
pub fn tokio_runtime() -> Result<tokio::runtime::Runtime, std::io::Error> {
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.thread_name_fn(|| {
static IDX: AtomicUsize = AtomicUsize::new(0);
let id = IDX.fetch_add(1, Ordering::Relaxed);
format!("tokio-{id}")
})
.build()
tokio::runtime::Builder::new_multi_thread().enable_all().build()
}

/// Runs the given future to completion or until a critical task panicked.
Expand Down
10 changes: 1 addition & 9 deletions crates/engine/tree/src/tree/payload_processor/executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,7 @@

use rayon::ThreadPool as RayonPool;
use std::{
sync::{
atomic::{AtomicUsize, Ordering},
Arc, OnceLock,
},
sync::{Arc, OnceLock},
time::Duration,
};
use tokio::{
Expand Down Expand Up @@ -74,7 +71,6 @@ impl WorkloadExecutorInner {
Handle::try_current().unwrap_or_else(|_| {
// Create a new runtime if no runtime is available
static RT: OnceLock<Runtime> = OnceLock::new();
static THREAD_COUNTER: AtomicUsize = AtomicUsize::new(0);

let rt = RT.get_or_init(|| {
Builder::new_multi_thread()
Expand All @@ -86,10 +82,6 @@ impl WorkloadExecutorInner {
// new block, and instead reuse the existing
// threads.
.thread_keep_alive(Duration::from_secs(15))
.thread_name_fn(|| {
let id = THREAD_COUNTER.fetch_add(1, Ordering::Relaxed);
format!("tokio-payload-{id}")
})
.build()
.unwrap()
});
Expand Down
2 changes: 1 addition & 1 deletion crates/tasks/src/pool.rs
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ impl BlockingTaskPool {
/// If a different stack size or other parameters are needed, they can be configured via
/// [`rayon::ThreadPoolBuilder`] returned by [`Self::builder`].
pub fn build() -> Result<Self, rayon::ThreadPoolBuildError> {
Self::builder().thread_name(|i| format!("rayon-{i}")).build().map(Self::new)
Self::builder().build().map(Self::new)
}

/// Asynchronous wrapper around Rayon's
Expand Down
10 changes: 1 addition & 9 deletions crates/trie/parallel/src/root.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,7 @@ use reth_trie::{
use reth_trie_db::{DatabaseHashedCursorFactory, DatabaseTrieCursorFactory};
use std::{
collections::HashMap,
sync::{
atomic::{AtomicUsize, Ordering},
mpsc, Arc, OnceLock,
},
sync::{mpsc, Arc, OnceLock},
time::Duration,
};
use thiserror::Error;
Expand Down Expand Up @@ -286,18 +283,13 @@ fn get_runtime_handle() -> Handle {
Handle::try_current().unwrap_or_else(|_| {
// Create a new runtime if no runtime is available
static RT: OnceLock<Runtime> = OnceLock::new();
static THREAD_COUNTER: AtomicUsize = AtomicUsize::new(0);

let rt = RT.get_or_init(|| {
Builder::new_multi_thread()
// Keep the threads alive for at least the block time (12 seconds) plus buffer.
// This prevents the costly process of spawning new threads on every
// new block, and instead reuses the existing threads.
.thread_keep_alive(Duration::from_secs(15))
.thread_name_fn(|| {
let id = THREAD_COUNTER.fetch_add(1, Ordering::Relaxed);
format!("tokio-trie-{id}")
})
.build()
.expect("Failed to create tokio runtime")
});
Expand Down