gavindiaz/solana-streamer
1
0
Fork 0
mirror of https://github.com/0xfnzero/solana-streamer.git synced 2026-08-15 18:08:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: restructure event handling from trait objects to enum pattern

Browse Source 
- Convert UnifiedEvent from trait object (Box<dyn UnifiedEvent>) to concrete enum type
  - Remove match_event! macro in favor of native match expressions
  - Simplify event metadata access: event.event_type() -> event.metadata().event_type
  - Unify event callback signatures: Fn(Box<dyn UnifiedEvent>) -> Fn(UnifiedEvent)
  - Update all example code to use the new enum-based event system
  - Optimize type system to reduce runtime overhead and improve type safety

  Affected scope:
  - Core event parser and processor
  - All protocol event definitions (PumpFun, PumpSwap, Bonk, Raydium series, etc.)
  - gRPC and Shred streaming modules
  - All example code
This commit is contained in:
ysq
2025-10-10 18:19:44 +08:00
parent 77caa39f0c
commit 312a675a40
43 changed files with 1013 additions and 1839 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/streaming/common/event_processor.rs
+41 -262
View File
@@ -1,21 +1,16 @@
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use crossbeam_queue::SegQueue;
use solana_sdk::pubkey::Pubkey;
use crate::common::AnyResult;
use crate::streaming::common::BackpressureStrategy;
use crate::streaming::common::{
MetricsEventType, MetricsManager, StreamClientConfig as ClientConfig,
};
use crate::streaming::common::{MetricsEventType, StreamClientConfig as ClientConfig};
use crate::streaming::event_parser::common::filter::EventTypeFilter;
use crate::streaming::event_parser::core::account_event_parser::AccountEventParser;
use crate::streaming::event_parser::core::common_event_parser::CommonEventParser;
use crate::streaming::event_parser::core::event_parser::EventParser;
use crate::streaming::event_parser::{core::traits::UnifiedEvent, Protocol};
use crate::streaming::grpc::{BackpressureConfig, EventPretty};
use crate::streaming::grpc::{EventPretty, MetricsManager};
use crate::streaming::shred::TransactionWithSlot;
use once_cell::sync::OnceCell;
@@ -24,141 +19,54 @@ pub enum EventSource {
Shred,
}
/// High-performance Event processor using SegQueue for all strategies
/// High-performance Event processor
pub struct EventProcessor {
pub(crate) metrics_manager: MetricsManager,
pub(crate) config: ClientConfig,
pub(crate) parser_cache: OnceCell<Arc<EventParser>>,
pub(crate) protocols: Vec<Protocol>,
pub(crate) event_type_filter: Option<EventTypeFilter>,
pub(crate) callback: Option<Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>>,
pub(crate) backpressure_config: BackpressureConfig,
pub(crate) grpc_queue: Arc<SegQueue<(EventPretty, Option<Pubkey>)>>,
pub(crate) shred_queue: Arc<SegQueue<(TransactionWithSlot, Option<Pubkey>)>>,
pub(crate) grpc_pending_count: Arc<AtomicUsize>,
pub(crate) shred_pending_count: Arc<AtomicUsize>,
pub(crate) processing_shutdown: Arc<AtomicBool>,
pub(crate) callback: Option<Arc<dyn Fn(UnifiedEvent) + Send + Sync>>,
}
impl EventProcessor {
pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self {
let backpressure_config = config.backpressure.clone();
let grpc_queue = Arc::new(SegQueue::new());
let shred_queue = Arc::new(SegQueue::new());
let grpc_pending_count = Arc::new(AtomicUsize::new(0));
let shred_pending_count = Arc::new(AtomicUsize::new(0));
let processing_shutdown = Arc::new(AtomicBool::new(false));
pub fn new(config: ClientConfig) -> Self {
Self {
metrics_manager,
config,
parser_cache: OnceCell::new(),
protocols: vec![],
event_type_filter: None,
backpressure_config,
callback: None,
grpc_queue,
shred_queue,
grpc_pending_count,
shred_pending_count,
processing_shutdown,
}
}
pub fn set_protocols_and_event_type_filter(
&mut self,
source: EventSource,
_source: EventSource,
protocols: Vec<Protocol>,
event_type_filter: Option<EventTypeFilter>,
backpressure_config: BackpressureConfig,
callback: Option<Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>>,
callback: Option<Arc<dyn Fn(UnifiedEvent) + Send + Sync>>,
) {
self.protocols = protocols;
self.event_type_filter = event_type_filter;
self.backpressure_config = backpressure_config;
self.callback = callback;
let protocols_ref = &self.protocols;
let event_type_filter_ref = self.event_type_filter.as_ref();
self.parser_cache.get_or_init(|| {
Arc::new(EventParser::new(protocols_ref.clone(), event_type_filter_ref.cloned()))
});
if matches!(self.backpressure_config.strategy, BackpressureStrategy::Block) {
self.start_block_processing_thread(source);
}
}
pub fn get_parser(&self) -> Arc<EventParser> {
self.parser_cache.get().unwrap().clone()
}
fn create_adapter_callback(&self) -> Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync> {
let callback = self.callback.clone().unwrap();
let metrics_manager = self.metrics_manager.clone();
Arc::new(move |event: Box<dyn UnifiedEvent>| {
let processing_time_us = event.handle_us() as f64;
fn invoke_callback(&self, event: UnifiedEvent) {
if let Some(callback) = self.callback.as_ref() {
callback(event);
metrics_manager.update_metrics(MetricsEventType::Transaction, 1, processing_time_us);
})
}
pub async fn process_grpc_event_transaction_with_metrics(
&self,
event_pretty: EventPretty,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()> {
self.apply_backpressure_control(event_pretty, bot_wallet).await
}
async fn apply_backpressure_control(
&self,
event_pretty: EventPretty,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()> {
match self.backpressure_config.strategy {
BackpressureStrategy::Block => {
loop {
let current_pending = self.grpc_pending_count.load(Ordering::Relaxed);
if current_pending < self.backpressure_config.permits {
self.grpc_queue.push((event_pretty, bot_wallet));
self.grpc_pending_count.fetch_add(1, Ordering::Relaxed);
break;
}
tokio::task::yield_now().await;
}
Ok(())
}
BackpressureStrategy::Drop => {
let current_pending = self.grpc_pending_count.load(Ordering::Relaxed);
if current_pending >= self.backpressure_config.permits {
self.metrics_manager.increment_dropped_events();
Ok(())
} else {
self.grpc_pending_count.fetch_add(1, Ordering::Relaxed);
let processor = self.clone();
tokio::spawn(async move {
match processor
.process_grpc_event_transaction(event_pretty, bot_wallet)
.await
{
Ok(_) => {
processor.grpc_pending_count.fetch_sub(1, Ordering::Relaxed);
}
Err(e) => {
log::error!("Error in async gRPC processing: {}", e);
}
}
});
Ok(())
}
}
}
}
async fn process_grpc_event_transaction(
pub async fn process_grpc_transaction(
&self,
event_pretty: EventPretty,
bot_wallet: Option<Pubkey>,
@@ -168,20 +76,20 @@ impl EventProcessor {
}
match event_pretty {
EventPretty::Account(account_pretty) => {
self.metrics_manager.add_account_process_count();
MetricsManager::global().add_account_process_count();
let account_event = AccountEventParser::parse_account_event(
&self.protocols,
account_pretty,
self.event_type_filter.as_ref(),
);
if let Some(event) = account_event {
let processing_time_us = event.handle_us() as f64;
if let Some(mut event) = account_event {
let processing_time_us = event.metadata().handle_us as f64;
self.invoke_callback(event);
self.update_metrics(MetricsEventType::Account, 1, processing_time_us);
}
}
EventPretty::Transaction(transaction_pretty) => {
self.metrics_manager.add_tx_process_count();
MetricsManager::global().add_tx_process_count();
let slot = transaction_pretty.slot;
let signature = transaction_pretty.signature;
let block_time = transaction_pretty.block_time;
@@ -190,7 +98,17 @@ impl EventProcessor {
let grpc_tx = transaction_pretty.grpc_tx;
let parser = self.get_parser();
let adapter_callback = self.create_adapter_callback();
let callback = self.callback.clone().unwrap();
let adapter_callback = Arc::new(move |mut event: UnifiedEvent| {
let processing_time_us = event.metadata().handle_us as f64;
callback(event);
MetricsManager::global().update_metrics(
MetricsEventType::Transaction,
1,
processing_time_us,
);
});
parser
.parse_grpc_transaction_owned(
grpc_tx,
@@ -205,18 +123,18 @@ impl EventProcessor {
.await?;
}
EventPretty::BlockMeta(block_meta_pretty) => {
self.metrics_manager.add_block_meta_process_count();
MetricsManager::global().add_block_meta_process_count();
let block_time_ms = block_meta_pretty
.block_time
.map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000)
.unwrap_or_else(|| chrono::Utc::now().timestamp_millis());
let block_meta_event = CommonEventParser::generate_block_meta_event(
let mut block_meta_event = CommonEventParser::generate_block_meta_event(
block_meta_pretty.slot,
block_meta_pretty.block_hash,
block_time_ms,
block_meta_pretty.recv_us,
);
let processing_time_us = block_meta_event.handle_us() as f64;
let processing_time_us = block_meta_event.metadata().handle_us as f64;
self.invoke_callback(block_meta_event);
self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us);
}
@@ -225,74 +143,6 @@ impl EventProcessor {
Ok(())
}
pub fn invoke_callback(&self, event: Box<dyn UnifiedEvent>) {
if let Some(callback) = self.callback.as_ref() {
callback(event);
}
}
pub async fn process_shred_transaction_immediate(
&self,
transaction_with_slot: TransactionWithSlot,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()> {
self.process_shred_transaction(transaction_with_slot, bot_wallet).await
}
pub async fn process_shred_transaction_with_metrics(
&self,
transaction_with_slot: TransactionWithSlot,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()> {
self.apply_shred_backpressure_control(transaction_with_slot, bot_wallet).await
}
async fn apply_shred_backpressure_control(
&self,
transaction_with_slot: TransactionWithSlot,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()> {
match self.backpressure_config.strategy {
BackpressureStrategy::Block => {
loop {
let current_pending = self.shred_pending_count.load(Ordering::Relaxed);
if current_pending < self.backpressure_config.permits {
self.shred_queue.push((transaction_with_slot, bot_wallet));
self.shred_pending_count.fetch_add(1, Ordering::Relaxed);
break;
}
tokio::task::yield_now().await;
}
Ok(())
}
BackpressureStrategy::Drop => {
let current_pending = self.shred_pending_count.load(Ordering::Relaxed);
if current_pending >= self.backpressure_config.permits {
self.metrics_manager.increment_dropped_events();
Ok(())
} else {
self.shred_pending_count.fetch_add(1, Ordering::Relaxed);
let processor = self.clone();
tokio::spawn(async move {
match processor
.process_shred_transaction(transaction_with_slot, bot_wallet)
.await
{
Ok(_) => {
processor.shred_pending_count.fetch_sub(1, Ordering::Relaxed);
}
Err(e) => {
log::error!("Error in async shred processing: {}", e);
}
}
});
Ok(())
}
}
}
}
pub async fn process_shred_transaction(
&self,
transaction_with_slot: TransactionWithSlot,
@@ -301,7 +151,7 @@ impl EventProcessor {
if self.callback.is_none() {
return Ok(());
}
self.metrics_manager.add_tx_process_count();
MetricsManager::global().add_tx_process_count();
let tx = transaction_with_slot.transaction;
let slot = transaction_with_slot.slot;
@@ -312,7 +162,17 @@ impl EventProcessor {
let recv_us = transaction_with_slot.recv_us;
let parser = self.get_parser();
let adapter_callback = self.create_adapter_callback();
let callback = self.callback.clone().unwrap();
let adapter_callback = Arc::new(move |mut event: UnifiedEvent| {
let processing_time_us = event.metadata().handle_us as f64;
callback(event);
MetricsManager::global().update_metrics(
MetricsEventType::Transaction,
1,
processing_time_us,
);
});
parser
.parse_versioned_transaction_owned(
tx,
@@ -331,99 +191,18 @@ impl EventProcessor {
}
fn update_metrics(&self, ty: MetricsEventType, count: u64, time_us: f64) {
self.metrics_manager.update_metrics(ty, count, time_us);
}
fn start_block_processing_thread(&self, source: EventSource) {
self.processing_shutdown.store(false, Ordering::Relaxed);
let grpc_queue = Arc::clone(&self.grpc_queue);
let shred_queue = Arc::clone(&self.shred_queue);
let grpc_pending_count = Arc::clone(&self.grpc_pending_count);
let shred_pending_count = Arc::clone(&self.shred_pending_count);
let shutdown_flag = Arc::clone(&self.processing_shutdown);
let shutdown_flag_clone = Arc::clone(&self.processing_shutdown);
let processor = self.clone();
let processor_clone = self.clone();
// Dedicated thread with busy-wait and lock-free processing
match source {
EventSource::Grpc => {
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();
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));
}
}
});
}
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个线程
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);
}
} else {
// 待测试替换方案： lock-free queue + spin + batch
std::thread::sleep(std::time::Duration::from_micros(500));
}
}
});
}
}
}
pub fn stop_processing(&self) {
self.processing_shutdown.store(true, Ordering::Relaxed);
MetricsManager::global().update_metrics(ty, count, time_us);
}
}
impl Clone for EventProcessor {
fn clone(&self) -> Self {
Self {
metrics_manager: self.metrics_manager.clone(),
config: self.config.clone(),
parser_cache: self.parser_cache.clone(),
protocols: self.protocols.clone(),
event_type_filter: self.event_type_filter.clone(),
backpressure_config: self.backpressure_config.clone(),
callback: self.callback.clone(),
grpc_queue: self.grpc_queue.clone(),
shred_queue: self.shred_queue.clone(),
grpc_pending_count: self.grpc_pending_count.clone(),
shred_pending_count: self.shred_pending_count.clone(),
processing_shutdown: self.processing_shutdown.clone(),
}
}
}