run snapshot segments in parallel

crates/snapshot/Cargo.toml

@@ -26,6 +26,7 @@ tokio = { workspace = true, features = ["sync"] }
 thiserror.workspace = true
 tracing.workspace = true
 clap = { workspace = true, features = ["derive"], optional = true }
+rayon.workspace = true
 
 [dev-dependencies]
 # reth

crates/snapshot/src/segments/mod.rs

@@ -27,7 +27,7 @@ use std::{ops::RangeInclusive, path::Path, sync::Arc};
 pub(crate) type Rows<const COLUMNS: usize> = [Vec<Vec<u8>>; COLUMNS];
 
 /// A segment represents a snapshotting of some portion of the data.
-pub trait Segment<DB: Database> {
+pub trait Segment<DB: Database>: Send + Sync {
     /// Returns the [`SnapshotSegment`].
     fn segment(&self) -> SnapshotSegment;
 

crates/snapshot/src/snapshotter.rs

@@ -1,6 +1,7 @@
 //! Support for snapshotting.
 
 use crate::{segments, segments::Segment};
+use rayon::prelude::*;
 use reth_db::database::Database;
 use reth_interfaces::RethResult;
 use reth_primitives::{snapshot::HighestSnapshots, BlockNumber};
@@ -98,18 +99,20 @@ impl<DB: Database> Snapshotter<DB> {
             segments.push((Box::new(segments::Receipts), block_range));
         }
 
-        for (segment, block_range) in &segments {
+        segments.par_iter().try_for_each(|(segment, block_range)| -> RethResult<()> {
             debug!(target: "snapshot", segment = %segment.segment(), ?block_range, "Snapshotting segment");
             let start = Instant::now();
 
             // Create a new database transaction on every segment to prevent long-lived read-only
             // transactions
-            let provider = self.provider_factory.provider()?;
+            let provider = self.provider_factory.provider()?.disable_long_read_transaction_safety();
             segment.snapshot(provider, self.snapshot_provider.clone(), block_range.clone())?;
 
             let elapsed = start.elapsed(); // TODO(alexey): track in metrics
             debug!(target: "snapshot", segment = %segment.segment(), ?block_range, ?elapsed, "Finished snapshotting segment");
-        }
+
+            Ok(())
+        })?;
 
         self.snapshot_provider.commit()?;
         for (segment, block_range) in segments {