From 5acf69714df0989513a73312014f97f366f2dfd2 Mon Sep 17 00:00:00 2001 From: ysq Date: Fri, 19 Sep 2025 23:17:50 +0800 Subject: [PATCH] feat: add EventSource enum and optimize event processing threads - Introduce EventSource enum (Grpc, Shred) for better event source management - Refactor start_block_processing_thread to conditionally spawn threads based on source type - Update callers to specify event source type --- src/streaming/common/event_processor.rs | 99 ++++++++++++++----------- src/streaming/shred_stream.rs | 1 + src/streaming/yellowstone_grpc.rs | 1 + 3 files changed, 58 insertions(+), 43 deletions(-) diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 19ce5b1..1504180 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -19,6 +19,11 @@ use crate::streaming::grpc::{BackpressureConfig, EventPretty}; use crate::streaming::shred::TransactionWithSlot; use once_cell::sync::OnceCell; +pub enum EventSource { + Grpc, + Shred, +} + /// High-performance Event processor using SegQueue for all strategies pub struct EventProcessor { pub(crate) metrics_manager: MetricsManager, @@ -62,6 +67,7 @@ impl EventProcessor { pub fn set_protocols_and_event_type_filter( &mut self, + source: EventSource, protocols: Vec, event_type_filter: Option, backpressure_config: BackpressureConfig, @@ -79,7 +85,7 @@ impl EventProcessor { }); if matches!(self.backpressure_config.strategy, BackpressureStrategy::Block) { - self.start_block_processing_thread(); + self.start_block_processing_thread(source); } } @@ -328,7 +334,7 @@ impl EventProcessor { self.metrics_manager.update_metrics(ty, count, time_us); } - fn start_block_processing_thread(&self) { + fn start_block_processing_thread(&self, source: EventSource) { self.processing_shutdown.store(false, Ordering::Relaxed); let grpc_queue = Arc::clone(&self.grpc_queue); @@ -340,55 +346,62 @@ impl EventProcessor { let processor = self.clone(); let processor_clone = self.clone(); // Dedicated thread with busy-wait and lock-free processing - std::thread::spawn(move || { - let worker_threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); // 如果获取失败则回退到4个线程 + match source { + EventSource::Grpc => { + std::thread::spawn(move || { + let mut worker_threads = + std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); // 如果获取失败则回退到4个线程 - let rt = tokio::runtime::Builder::new_multi_thread() - .worker_threads(worker_threads) - .enable_all() - .build() - .unwrap(); + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(worker_threads) + .enable_all() + .build() + .unwrap(); - while !shutdown_flag.load(Ordering::Relaxed) { - if let Some((event_pretty, bot_wallet)) = grpc_queue.pop() { - grpc_pending_count.fetch_sub(1, Ordering::Relaxed); - if let Err(e) = rt.block_on( - processor.process_grpc_event_transaction(event_pretty, bot_wallet), - ) { - println!("Error processing gRPC event: {}", e); + while !shutdown_flag.load(Ordering::Relaxed) { + if let Some((event_pretty, bot_wallet)) = grpc_queue.pop() { + grpc_pending_count.fetch_sub(1, Ordering::Relaxed); + if let Err(e) = rt.block_on( + processor.process_grpc_event_transaction(event_pretty, bot_wallet), + ) { + println!("Error processing gRPC event: {}", e); + } + } else { + // 待测试替换方案： lock-free queue + spin + batch + std::thread::sleep(std::time::Duration::from_micros(500)); + } } - } else { - // Yield to reduce CPU usage in busy wait - std::thread::yield_now(); - } + }); } - }); + EventSource::Shred => { + // Shred processing with same low-latency optimization + std::thread::spawn(move || { + let worker_threads = + std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); // 如果获取失败则回退到4个线程 - // Shred processing with same low-latency optimization - std::thread::spawn(move || { - let worker_threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); // 如果获取失败则回退到4个线程 + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(worker_threads) + .enable_all() + .build() + .unwrap(); - let rt = tokio::runtime::Builder::new_multi_thread() - .worker_threads(worker_threads) - .enable_all() - .build() - .unwrap(); - - while !shutdown_flag_clone.load(Ordering::Relaxed) { - if let Some((transaction_with_slot, bot_wallet)) = shred_queue.pop() { - shred_pending_count.fetch_sub(1, Ordering::Relaxed); - if let Err(e) = rt.block_on( - processor_clone - .process_shred_transaction(transaction_with_slot, bot_wallet), - ) { - log::error!("Error processing shred transaction: {}", e); + while !shutdown_flag_clone.load(Ordering::Relaxed) { + if let Some((transaction_with_slot, bot_wallet)) = shred_queue.pop() { + shred_pending_count.fetch_sub(1, Ordering::Relaxed); + if let Err(e) = rt.block_on( + processor_clone + .process_shred_transaction(transaction_with_slot, bot_wallet), + ) { + log::error!("Error processing shred transaction: {}", e); + } + } else { + // 待测试替换方案： lock-free queue + spin + batch + std::thread::sleep(std::time::Duration::from_micros(500)); + } } - } else { - // Yield to reduce CPU usage in busy wait - std::thread::yield_now(); - } + }); } - }); + } } pub fn stop_processing(&self) { diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 2cb6a49..4aa7bd5 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -40,6 +40,7 @@ impl ShredStreamGrpc { let mut event_processor = EventProcessor::new(self.metrics_manager.clone(), self.config.clone()); event_processor.set_protocols_and_event_type_filter( + super::common::EventSource::Shred, protocols, event_type_filter, self.config.backpressure.clone(), diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 12ac830..ebd73a6 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -210,6 +210,7 @@ impl YellowstoneGrpc { // 启动流处理任务 let mut event_processor = self.event_processor.clone(); event_processor.set_protocols_and_event_type_filter( + super::common::EventSource::Grpc, protocols, event_type_filter, self.config.backpressure.clone(),