From 7601eb370cfbe1fda97a8fbbed061d6c477e6f83 Mon Sep 17 00:00:00 2001 From: ysq Date: Wed, 3 Sep 2025 15:29:57 +0800 Subject: [PATCH] perf: optimize event processing and add gRPC instruction parsing --- src/streaming/common/event_processor.rs | 73 ++-- src/streaming/event_parser/common/types.rs | 167 ++++++++ src/streaming/event_parser/core/macros.rs | 37 ++ src/streaming/event_parser/core/traits.rs | 425 ++++++++++++++++++--- src/streaming/grpc/pool.rs | 3 +- src/streaming/grpc/types.rs | 14 +- src/streaming/yellowstone_sub_system.rs | 28 +- 7 files changed, 627 insertions(+), 120 deletions(-) diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 300dc22..c25930f 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -1,7 +1,6 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; - use crossbeam_queue::SegQueue; use solana_sdk::pubkey::Pubkey; @@ -31,19 +30,14 @@ pub struct EventProcessor { pub(crate) event_type_filter: Option, pub(crate) callback: Option) + Send + Sync>>, pub(crate) backpressure_config: BackpressureConfig, - /// High-performance lockfree queue for gRPC events pub(crate) grpc_queue: Arc)>>, - /// High-performance lockfree queue for shred events pub(crate) shred_queue: Arc)>>, - /// Fast O(1) counter for Drop strategy (avoids expensive SegQueue::len()) pub(crate) grpc_pending_count: Arc, pub(crate) shred_pending_count: Arc, - /// Processing thread control pub(crate) processing_shutdown: Arc, } impl EventProcessor { - /// Create a new high-performance event processor pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self { let backpressure_config = config.backpressure.clone(); let grpc_queue = Arc::new(SegQueue::new()); @@ -78,21 +72,15 @@ impl EventProcessor { self.protocols = protocols; self.event_type_filter = event_type_filter; - // Check if Block processing thread should be started (before moving backpressure_config) - let should_start_block_processing = true; - // matches!(backpressure_config.strategy, BackpressureStrategy::Block); - self.backpressure_config = backpressure_config; self.callback = callback; - // Use stored values to initialize parser_cache let protocols_ref = &self.protocols; let event_type_filter_ref = self.event_type_filter.as_ref(); self.parser_cache.get_or_init(|| { Arc::new(MutilEventParser::new(protocols_ref.clone(), event_type_filter_ref.cloned())) }); - // Start Block processing thread if using Block strategy - if should_start_block_processing { + if matches!(self.backpressure_config.strategy, BackpressureStrategy::Block) { self.start_block_processing_thread(); } } @@ -101,7 +89,6 @@ impl EventProcessor { self.parser_cache.get().unwrap().clone() } - /// Create adapter callback fn create_adapter_callback(&self) -> Arc) + Send + Sync> { let callback = self.callback.clone().unwrap(); let metrics_manager = self.metrics_manager.clone(); @@ -121,7 +108,6 @@ impl EventProcessor { self.apply_backpressure_control(event_pretty, bot_wallet).await } - /// Apply backpressure control strategy async fn apply_backpressure_control( &self, event_pretty: EventPretty, @@ -129,7 +115,6 @@ impl EventProcessor { ) -> AnyResult<()> { match self.backpressure_config.strategy { BackpressureStrategy::Block => { - // Block strategy: async wait if queue is full (backpressure control) loop { let current_pending = self.grpc_pending_count.load(Ordering::Relaxed); if current_pending < self.backpressure_config.permits { @@ -137,14 +122,12 @@ impl EventProcessor { self.grpc_pending_count.fetch_add(1, Ordering::Relaxed); break; } - // Async yield to avoid blocking gRPC data source + tokio::task::yield_now().await; } Ok(()) } BackpressureStrategy::Drop => { - // Drop strategy: Use O(1) atomic counter instead of expensive O(n) len() - // If pending count >= permits, DROP the event immediately let current_pending = self.grpc_pending_count.load(Ordering::Relaxed); if current_pending >= self.backpressure_config.permits { self.metrics_manager.increment_dropped_events(); @@ -197,16 +180,16 @@ impl EventProcessor { self.metrics_manager.add_tx_process_count(); let slot = transaction_pretty.slot; let signature = transaction_pretty.signature; - let tx = transaction_pretty.tx; let block_time = transaction_pretty.block_time; let program_received_time_us = transaction_pretty.program_received_time_us; let transaction_index = transaction_pretty.transaction_index; - // Use cache to get parser + let grpc_tx = transaction_pretty.grpc_tx; + let parser = self.get_parser(); let adapter_callback = self.create_adapter_callback(); parser - .parse_transaction_owned( - tx, + .parse_grpc_transaction_owned( + grpc_tx, signature, Some(slot), block_time, @@ -244,7 +227,6 @@ impl EventProcessor { } } - /// Process a single transaction immediately pub async fn process_shred_transaction_immediate( &self, transaction_with_slot: TransactionWithSlot, @@ -253,17 +235,14 @@ impl EventProcessor { self.process_shred_transaction(transaction_with_slot, bot_wallet).await } - /// Process shred transaction with backpressure control and performance monitoring pub async fn process_shred_transaction_with_metrics( &self, transaction_with_slot: TransactionWithSlot, bot_wallet: Option, ) -> AnyResult<()> { - // Backpressure control logic self.apply_shred_backpressure_control(transaction_with_slot, bot_wallet).await } - /// Apply shred backpressure control strategy async fn apply_shred_backpressure_control( &self, transaction_with_slot: TransactionWithSlot, @@ -271,7 +250,6 @@ impl EventProcessor { ) -> AnyResult<()> { match self.backpressure_config.strategy { BackpressureStrategy::Block => { - // Block strategy: async wait if queue is full (backpressure control) loop { let current_pending = self.shred_pending_count.load(Ordering::Relaxed); if current_pending < self.backpressure_config.permits { @@ -279,13 +257,12 @@ impl EventProcessor { self.shred_pending_count.fetch_add(1, Ordering::Relaxed); break; } - // Async yield to avoid blocking shred data source + tokio::task::yield_now().await; } Ok(()) } BackpressureStrategy::Drop => { - // Drop strategy: Use O(1) atomic counter instead of expensive O(n) len() let current_pending = self.shred_pending_count.load(Ordering::Relaxed); if current_pending >= self.backpressure_config.permits { self.metrics_manager.increment_dropped_events(); @@ -326,7 +303,7 @@ impl EventProcessor { let slot = transaction_with_slot.slot; let signature = tx.signatures[0]; let program_received_time_us = transaction_with_slot.program_received_time_us; - // Use cache to get parser + let parser = self.get_parser(); let adapter_callback = self.create_adapter_callback(); parser @@ -350,9 +327,7 @@ impl EventProcessor { self.metrics_manager.update_metrics(ty, count, time_us); } - /// Start dedicated processing threads for all strategies fn start_block_processing_thread(&self) { - // Reset shutdown flag self.processing_shutdown.store(false, Ordering::Relaxed); let grpc_queue = Arc::clone(&self.grpc_queue); @@ -363,36 +338,44 @@ impl EventProcessor { let shutdown_flag_clone = Arc::clone(&self.processing_shutdown); let processor = self.clone(); let processor_clone = self.clone(); - // 1. 专用线程 + 2. Busy-wait + 4. 无锁处理 + // Dedicated thread with busy-wait and lock-free processing std::thread::spawn(move || { - // 创建blocking runtime for async processing - let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + 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() { - // Decrement pending counter when consuming from queue grpc_pending_count.fetch_sub(1, Ordering::Relaxed); - // Process event in blocking runtime if let Err(e) = rt.block_on( processor.process_grpc_event_transaction(event_pretty, bot_wallet), ) { println!("Error processing gRPC event: {}", e); } } else { - // 2. 优化忙等待: 使用轻量级休眠减少CPU占用 + // Yield to reduce CPU usage in busy wait std::thread::yield_now(); } } }); - // Shred处理也使用相同的低延迟优化 + // Shred processing with same low-latency optimization std::thread::spawn(move || { - let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + 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() { - // Decrement pending counter when consuming from queue shred_pending_count.fetch_sub(1, Ordering::Relaxed); - // Process transaction in blocking runtime if let Err(e) = rt.block_on( processor_clone .process_shred_transaction(transaction_with_slot, bot_wallet), @@ -400,20 +383,18 @@ impl EventProcessor { log::error!("Error processing shred transaction: {}", e); } } else { - // 优化忙等待: 使用轻量级休眠减少CPU占用 + // Yield to reduce CPU usage in busy wait std::thread::yield_now(); } } }); } - /// Stop processing threads pub fn stop_processing(&self) { self.processing_shutdown.store(true, Ordering::Relaxed); } } -// Implement Clone trait to support sharing between modules impl Clone for EventProcessor { fn clone(&self) -> Self { Self { diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 2daa630..877e940 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -519,3 +519,170 @@ pub fn parse_swap_data_from_next_instructions( None } } + +/// Parse token transfer data from next instructions +/// TODO: - wait refactor +pub fn parse_swap_data_from_next_grpc_instructions( + event: &dyn UnifiedEvent, + inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstructions, + current_index: i8, + accounts: &[Pubkey], +) -> Option { + let mut swap_data = SwapData { + from_mint: Pubkey::default(), + to_mint: Pubkey::default(), + from_amount: 0, + to_amount: 0, + description: None, + }; + + // 先根据 event 取出关键信息 + let mut user: Option = None; + let mut from_mint: Option = None; + let mut to_mint: Option = None; + let mut user_from_token: Option = None; + let mut user_to_token: Option = None; + let mut from_vault: Option = None; + let mut to_vault: Option = None; + + match_event!(&*event, { + BonkTradeEvent => |e: BonkTradeEvent| { + user = Some(e.payer); + from_mint = Some(e.base_token_mint); + to_mint = Some(e.quote_token_mint); + user_from_token = Some(e.user_base_token); + user_to_token = Some(e.user_quote_token); + from_vault = Some(e.base_vault); + to_vault = Some(e.quote_vault); + }, + PumpFunTradeEvent => |e: PumpFunTradeEvent| { + swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint }; + swap_data.to_mint = if e.is_buy { e.mint } else { *SOL_MINT }; + }, + PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { + swap_data.from_mint = e.quote_mint; + swap_data.to_mint = e.base_mint; + }, + PumpSwapSellEvent => |e: PumpSwapSellEvent| { + swap_data.from_mint = e.base_mint; + swap_data.to_mint = e.quote_mint; + }, + RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { + user = Some(e.payer); + from_mint = Some(e.input_token_mint); + to_mint = Some(e.output_token_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { + user = Some(e.payer); + swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into()); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { + user = Some(e.payer); + from_mint = Some(e.input_vault_mint); + to_mint = Some(e.output_vault_mint); + user_from_token = Some(e.input_token_account); + user_to_token = Some(e.output_token_account); + from_vault = Some(e.input_vault); + to_vault = Some(e.output_vault); + }, + RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { + user = Some(e.user_source_owner); + swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into()); + user_from_token = Some(e.user_source_token_account); + user_to_token = Some(e.user_destination_token_account); + from_vault = Some(e.pool_pc_token_account); + to_vault = Some(e.pool_coin_token_account); + }, + }); + + let user_to_token = user_to_token.unwrap_or_default(); + let user_from_token = user_from_token.unwrap_or_default(); + let to_vault = to_vault.unwrap_or_default(); + let from_vault = from_vault.unwrap_or_default(); + let to_mint = to_mint.unwrap_or_default(); + let from_mint = from_mint.unwrap_or_default(); + + // 单次循环完成提取和判断 + for instruction in inner_instruction.instructions.iter().skip((current_index + 1) as usize) { + let compiled = &instruction; + let program_id = accounts[compiled.program_id_index as usize]; + if !SYSTEM_PROGRAMS.contains(&program_id) { + break; + } + let data = &compiled.data; + + // 使用 SIMD 验证数据格式 + if !SimdUtils::validate_data_format(data, 8) { + continue; + } + + let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize]; + let (source, destination, amount) = match data[0] { + 12 if compiled.accounts.len() >= 4 => { + let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); + (get_pubkey(0), get_pubkey(2), amt) + } + 3 if compiled.accounts.len() >= 3 => { + let amt = u64::from_le_bytes(data[1..9].try_into().unwrap()); + (get_pubkey(0), get_pubkey(1), amt) + } + 2 if compiled.accounts.len() >= 2 => { + let amt = u64::from_le_bytes(data[4..12].try_into().unwrap()); + (get_pubkey(0), get_pubkey(1), amt) + } + _ => continue, + }; + + match (source, destination) { + (s, d) if s == user_to_token && d == to_vault => { + swap_data.from_mint = to_mint; + swap_data.from_amount = amount; + } + (s, d) if s == from_vault && d == user_from_token => { + swap_data.to_mint = from_mint; + swap_data.to_amount = amount; + } + (s, d) if s == user_from_token && d == from_vault => { + swap_data.from_mint = from_mint; + swap_data.from_amount = amount; + } + (s, d) if s == to_vault && d == user_to_token => { + swap_data.to_mint = to_mint; + swap_data.to_amount = amount; + } + (s, d) if s == user_from_token && d == to_vault => { + swap_data.from_mint = from_mint; + swap_data.from_amount = amount; + } + (s, d) if s == from_vault && d == user_to_token => { + swap_data.to_mint = to_mint; + swap_data.to_amount = amount; + } + _ => {} + } + if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() { + break; + } + if swap_data.from_amount != 0 && swap_data.to_amount != 0 { + break; + } + } + + if swap_data.from_mint != Pubkey::default() + || swap_data.to_mint != Pubkey::default() + || swap_data.from_amount != 0 + || swap_data.to_amount != 0 + { + Some(swap_data) + } else { + None + } +} diff --git a/src/streaming/event_parser/core/macros.rs b/src/streaming/event_parser/core/macros.rs index 724ccc0..12dd7e2 100644 --- a/src/streaming/event_parser/core/macros.rs +++ b/src/streaming/event_parser/core/macros.rs @@ -49,6 +49,21 @@ macro_rules! impl_event_parser_delegate { ) } + fn parse_events_from_grpc_inner_instruction( + &self, + inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction, + signature: solana_sdk::signature::Signature, + slot: u64, + block_time: Option, + program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + transaction_index: Option, + config: &GenericEventParseConfig, + ) -> Vec> { + self.inner.parse_events_from_grpc_inner_instruction(inner_instruction, signature, slot, block_time, program_received_time_us, outer_index, inner_index, transaction_index, config) + } + fn parse_events_from_instruction( &self, instruction: &solana_sdk::instruction::CompiledInstruction, @@ -84,6 +99,28 @@ macro_rules! impl_event_parser_delegate { ) } + fn parse_events_from_grpc_instruction( + &self, + instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction, + accounts: &[solana_sdk::pubkey::Pubkey], + signature: solana_sdk::signature::Signature, + slot: u64, + block_time: Option, + program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + bot_wallet: Option, + transaction_index: Option, + inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, + callback: std::sync::Arc< + dyn for<'a> Fn(&'a Box) + + Send + + Sync, + >, + ) -> anyhow::Result<()> { + self.inner.parse_events_from_grpc_instruction(instruction, accounts, signature, slot, block_time, program_received_time_us, outer_index, inner_index, bot_wallet, transaction_index, inner_instructions, callback) + } + fn should_handle(&self, program_id: &solana_sdk::pubkey::Pubkey) -> bool { self.inner.should_handle(program_id) } diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index d45e2dd..8106a78 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -13,13 +13,16 @@ use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; use std::time::Instant; +use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo; use super::global_state::{ add_bonk_dev_address, add_dev_address, is_bonk_dev_address, is_dev_address, }; use crate::streaming::common::simd_utils::SimdUtils; -use crate::streaming::event_parser::common::{parse_swap_data_from_next_instructions, SwapData}; +use crate::streaming::event_parser::common::{ + parse_swap_data_from_next_grpc_instructions, parse_swap_data_from_next_instructions, SwapData, +}; use crate::streaming::event_parser::protocols::pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}; use crate::streaming::event_parser::{ common::{EventMetadata, EventType, ProtocolType}, @@ -289,6 +292,21 @@ pub trait EventParser: Send + Sync { config: &GenericEventParseConfig, ) -> Vec>; + /// 从内联指令中解析事件数据 + #[allow(clippy::too_many_arguments)] + fn parse_events_from_grpc_inner_instruction( + &self, + inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction, + signature: Signature, + slot: u64, + block_time: Option, + program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + transaction_index: Option, + config: &GenericEventParseConfig, + ) -> Vec>; + /// 从指令中解析事件数据 #[allow(clippy::too_many_arguments)] fn parse_events_from_instruction( @@ -307,6 +325,80 @@ pub trait EventParser: Send + Sync { callback: Arc Fn(&'a Box) + Send + Sync>, ) -> anyhow::Result<()>; + /// 从指令中解析事件数据 + /// TODO: - wait refactor + #[allow(clippy::too_many_arguments)] + fn parse_events_from_grpc_instruction( + &self, + instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction, + accounts: &[Pubkey], + signature: Signature, + slot: u64, + block_time: Option, + program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + bot_wallet: Option, + transaction_index: Option, + inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()>; + + #[allow(clippy::too_many_arguments)] + async fn parse_instruction_events_from_grpc_transaction( + &self, + compiled_instructions: &[yellowstone_grpc_proto::prelude::CompiledInstruction], + signature: Signature, + slot: Option, + block_time: Option, + program_received_time_us: i64, + accounts: &[Pubkey], + inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions], + bot_wallet: Option, + transaction_index: Option, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { + // 获取交易的指令和账户 + let mut accounts = accounts.to_vec(); + // 检查交易中是否包含程序 + let has_program = accounts.iter().any(|account| self.should_handle(account)); + if has_program { + // 解析每个指令 + for (index, instruction) in compiled_instructions.iter().enumerate() { + if let Some(program_id) = accounts.get(instruction.program_id_index as usize) { + if self.should_handle(program_id) { + let max_idx = instruction.accounts.iter().max().unwrap_or(&0); + // 补齐accounts(使用Pubkey::default()) + if *max_idx as usize > accounts.len() { + for _i in accounts.len()..*max_idx as usize { + accounts.push(Pubkey::default()); + } + } + let inner_instructions = inner_instructions + .iter() + .find(|inner_instruction| inner_instruction.index == index as u32); + self.parse_grpc_instruction( + instruction, + &accounts, + signature, + slot, + block_time, + program_received_time_us, + index as i64, + None, + bot_wallet, + transaction_index, + inner_instructions, + Arc::clone(&callback), + ) + .await?; + } + } + } + } + Ok(()) + } + /// 从VersionedTransaction中解析指令事件的通用方法 #[allow(clippy::too_many_arguments)] async fn parse_instruction_events_from_versioned_transaction( @@ -424,10 +516,9 @@ pub trait EventParser: Send + Sync { Ok(()) } - /// 解析交易，使用所有权语义的回调以避免不必要的克隆 - async fn parse_transaction_owned( + async fn parse_grpc_transaction_owned( &self, - tx: TransactionWithStatusMeta, + grpc_tx: SubscribeUpdateTransactionInfo, signature: Signature, slot: Option, block_time: Option, @@ -441,8 +532,8 @@ pub trait EventParser: Send + Sync { callback(event.clone_boxed()); }); // 调用原始方法 - self.parse_transaction( - tx, + self.parse_grpc_transaction( + grpc_tx, signature, slot, block_time, @@ -454,9 +545,9 @@ pub trait EventParser: Send + Sync { .await } - async fn parse_transaction( + async fn parse_grpc_transaction( &self, - tx: TransactionWithStatusMeta, + grpc_tx: SubscribeUpdateTransactionInfo, signature: Signature, slot: Option, block_time: Option, @@ -465,60 +556,86 @@ pub trait EventParser: Send + Sync { transaction_index: Option, callback: Arc Fn(&'a Box) + Send + Sync>, ) -> anyhow::Result<()> { - let versioned_tx = tx.get_transaction(); - let meta = tx.get_status_meta(); - let mut address_table_lookups: Vec = vec![]; - let mut inner_instructions: Vec = vec![]; - if let Some(meta) = meta { - inner_instructions = meta.inner_instructions.unwrap_or_default(); - address_table_lookups.reserve( - meta.loaded_addresses.writable.len() + meta.loaded_addresses.readonly.len(), - ); - address_table_lookups.extend( - meta.loaded_addresses.writable.into_iter().chain(meta.loaded_addresses.readonly), - ); - } - let mut accounts = Vec::with_capacity( - versioned_tx.message.static_account_keys().len() + address_table_lookups.len(), - ); - accounts.extend_from_slice(versioned_tx.message.static_account_keys()); - accounts.extend(address_table_lookups); - // 使用 Arc 包装共享数据，避免不必要的克隆 - let accounts_arc = Arc::new(accounts); - let inner_instructions_arc = Arc::new(inner_instructions); - // 解析指令事件 - self.parse_instruction_events_from_versioned_transaction( - &versioned_tx, - signature, - slot, - block_time, - program_received_time_us, - &accounts_arc, - &inner_instructions_arc, - bot_wallet, - transaction_index, - callback.clone(), - ) - .await?; + if let Some(transition) = grpc_tx.transaction { + if let Some(message) = &transition.message { + let mut address_table_lookups: Vec> = vec![]; + let mut inner_instructions: Vec< + yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions, + > = vec![]; - // 解析嵌套指令事件 - for inner_instruction in inner_instructions_arc.iter() { - for (index, instruction) in inner_instruction.instructions.iter().enumerate() { - self.parse_instruction( - &instruction.instruction, - &accounts_arc, + if let Some(meta) = grpc_tx.meta { + inner_instructions = meta.inner_instructions; + address_table_lookups.reserve( + meta.loaded_writable_addresses.len() + meta.loaded_writable_addresses.len(), + ); + let loaded_writable_addresses = meta.loaded_writable_addresses; + let loaded_readonly_addresses = meta.loaded_readonly_addresses; + address_table_lookups.extend( + loaded_writable_addresses.into_iter().chain(loaded_readonly_addresses), + ); + } + + let mut accounts_bytes: Vec> = + Vec::with_capacity(message.account_keys.len() + address_table_lookups.len()); + accounts_bytes.extend_from_slice(&message.account_keys); + accounts_bytes.extend(address_table_lookups); + // 转换为 Pubkey + let accounts: Vec = accounts_bytes + .iter() + .filter_map(|account| { + if account.len() == 32 { + Some(Pubkey::try_from(account.as_slice()).unwrap_or_default()) + } else { + None + } + }) + .collect(); + // 使用 Arc 包装共享数据，避免不必要的克隆 + let accounts_arc = Arc::new(accounts); + let inner_instructions_arc = Arc::new(inner_instructions); + // 解析指令事件 + let instructions = &message.instructions; + self.parse_instruction_events_from_grpc_transaction( + &instructions, signature, slot, block_time, program_received_time_us, - inner_instruction.index as i64, - Some(index as i64), + &accounts_arc, + &inner_instructions_arc, bot_wallet, transaction_index, - Some(&inner_instruction), callback.clone(), ) .await?; + + // 解析嵌套指令事件 + for inner_instruction in inner_instructions_arc.iter() { + for (index, instruction) in inner_instruction.instructions.iter().enumerate() { + let accounts = &instruction.accounts; + let data = &instruction.data; + let instruction = yellowstone_grpc_proto::prelude::CompiledInstruction { + program_id_index: instruction.program_id_index, + accounts: accounts.to_vec(), + data: data.to_vec(), + }; + self.parse_grpc_instruction( + &instruction, + &accounts_arc, + signature, + slot, + block_time, + program_received_time_us, + inner_instruction.index as i64, + Some(index as i64), + bot_wallet, + transaction_index, + Some(&inner_instruction), + callback.clone(), + ) + .await?; + } + } } } @@ -684,6 +801,39 @@ pub trait EventParser: Send + Sync { Ok(events) } + #[allow(clippy::too_many_arguments)] + async fn parse_grpc_instruction( + &self, + instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction, + accounts: &[Pubkey], + signature: Signature, + slot: Option, + block_time: Option, + program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + bot_wallet: Option, + transaction_index: Option, + inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { + let slot = slot.unwrap_or(0); + self.parse_events_from_grpc_instruction( + instruction, + accounts, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + bot_wallet, + transaction_index, + inner_instructions, + callback, + ) + } + #[allow(clippy::too_many_arguments)] async fn parse_instruction( &self, @@ -894,6 +1044,42 @@ impl EventParser for GenericEventParser { events } + /// 从内联指令中解析事件数据 + #[allow(clippy::too_many_arguments)] + fn parse_events_from_grpc_inner_instruction( + &self, + inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction, + signature: Signature, + slot: u64, + block_time: Option, + program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + transaction_index: Option, + config: &GenericEventParseConfig, + ) -> Vec> { + // Use SIMD-optimized data validation + if !SimdUtils::validate_instruction_data_simd(&inner_instruction.data, 16, 0) { + return Vec::new(); + } + let data = &inner_instruction.data[16..]; + let mut events = Vec::new(); + if let Some(event) = self.parse_inner_instruction_event( + config, + data, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + transaction_index, + ) { + events.push(event); + } + events + } + /// 从指令中解析事件 #[allow(clippy::too_many_arguments)] fn parse_events_from_instruction( @@ -1028,6 +1214,141 @@ impl EventParser for GenericEventParser { Ok(()) } + /// 从指令中解析事件 + /// TODO: - wait refactor + #[allow(clippy::too_many_arguments)] + fn parse_events_from_grpc_instruction( + &self, + instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction, + accounts: &[Pubkey], + signature: Signature, + slot: u64, + block_time: Option, + program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + bot_wallet: Option, + transaction_index: Option, + inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>, + callback: Arc Fn(&'a Box) + Send + Sync>, + ) -> anyhow::Result<()> { + let program_id = accounts[instruction.program_id_index as usize]; + if !self.should_handle(&program_id) { + return Ok(()); + } + // 一维化并行处理：将所有 (discriminator, config) 组合展开并行处理 + let all_processing_params: Vec<_> = self + .instruction_configs + .iter() + .filter(|(disc, _)| { + // Use SIMD-optimized data validation and discriminator matching + SimdUtils::validate_instruction_data_simd(&instruction.data, disc.len(), disc.len()) + && SimdUtils::fast_discriminator_match(&instruction.data, disc) + }) + .flat_map(|(disc, configs)| { + configs + .iter() + .filter(|config| config.program_id == program_id) + .map(move |config| (disc, config)) + }) + .collect(); + + // Use SIMD-optimized account indices validation (只需检查一次) + if !SimdUtils::validate_account_indices_simd(&instruction.accounts, accounts.len()) { + return Ok(()); + } + + // 使用缓存构建账户公钥列表，避免重复分配 (只需构建一次) + let account_pubkeys = { + let mut cache_guard = self.account_cache.lock(); + cache_guard.build_account_pubkeys(&instruction.accounts, accounts).to_vec() + }; + + // 并行处理所有 (discriminator, config) 组合 + let all_results: Vec<_> = all_processing_params + .iter() + .filter_map(|(disc, config)| { + let data = &instruction.data[disc.len()..]; + self.parse_instruction_event( + config, + data, + &account_pubkeys, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + transaction_index, + ) + .map(|event| ((*disc).clone(), (*config).clone(), event)) + }) + .collect(); + + for (_disc, config, mut event) in all_results { + // 阻塞处理：原有的同步逻辑 + let mut inner_instruction_event: Option> = None; + if inner_instructions.is_some() { + let inner_instructions_ref = inner_instructions.unwrap(); + + // 并行执行两个任务 + let (inner_event_result, swap_data_result) = std::thread::scope(|s| { + let inner_event_handle = s.spawn(|| { + for inner_instruction in inner_instructions_ref.instructions.iter() { + let result = self.parse_events_from_grpc_inner_instruction( + &inner_instruction, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + transaction_index, + &config, + ); + if result.len() > 0 { + return Some(result[0].clone()); + } + } + None + }); + + let swap_data_handle = s.spawn(|| { + if !event.swap_data_is_parsed() { + parse_swap_data_from_next_grpc_instructions( + &*event, + inner_instructions_ref, + inner_index.unwrap_or(-1_i64) as i8, + &accounts, + ) + } else { + None + } + }); + + // 等待两个任务完成 + (inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap()) + }); + + inner_instruction_event = inner_event_result; + if let Some(swap_data) = swap_data_result { + event.set_swap_data(swap_data); + } + } + // 合并事件 + if let Some(inner_instruction_event) = inner_instruction_event { + event.merge(&*inner_instruction_event); + } + // 设置处理时间（使用高性能时钟） + event.set_program_handle_time_consuming_us(elapsed_micros_since( + program_received_time_us, + )); + event = process_event(event, bot_wallet); + callback(&event); + } + Ok(()) + } + fn should_handle(&self, program_id: &Pubkey) -> bool { self.program_ids.contains(program_id) } diff --git a/src/streaming/grpc/pool.rs b/src/streaming/grpc/pool.rs index 5371a54..afc2082 100644 --- a/src/streaming/grpc/pool.rs +++ b/src/streaming/grpc/pool.rs @@ -282,9 +282,8 @@ impl PooledTransactionPretty { self.transaction.signature = Signature::try_from(tx.signature.as_slice()).expect("valid signature"); self.transaction.is_vote = tx.is_vote; - self.transaction.tx = yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) - .expect("valid tx with meta"); self.transaction.program_received_time_us = get_high_perf_clock(); + self.transaction.grpc_tx = tx; } } diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index 1c450de..7f2e771 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -2,14 +2,17 @@ use solana_sdk::{pubkey::Pubkey, signature::Signature}; use solana_transaction_status::{TransactionWithStatusMeta, VersionedTransactionWithStatusMeta}; use std::{collections::HashMap, fmt}; use yellowstone_grpc_proto::{ - geyser::{SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions}, + geyser::{ + SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions, + SubscribeUpdateTransactionInfo, + }, prost_types::Timestamp, }; pub type TransactionsFilterMap = HashMap; pub type AccountsFilterMap = HashMap; -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum EventPretty { BlockMeta(BlockMetaPretty), Transaction(TransactionPretty), @@ -71,8 +74,8 @@ pub struct TransactionPretty { pub block_time: Option, pub signature: Signature, pub is_vote: bool, - pub tx: TransactionWithStatusMeta, pub program_received_time_us: i64, + pub grpc_tx: SubscribeUpdateTransactionInfo, } impl fmt::Debug for TransactionPretty { @@ -96,10 +99,7 @@ impl Default for TransactionPretty { block_time: None, signature: Signature::default(), is_vote: false, - tx: TransactionWithStatusMeta::Complete(VersionedTransactionWithStatusMeta { - transaction: solana_sdk::transaction::VersionedTransaction::default(), - meta: solana_transaction_status::TransactionStatusMeta::default(), - }), + grpc_tx: SubscribeUpdateTransactionInfo::default(), program_received_time_us: 0, } } diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 9d0c710..b572f2d 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -100,20 +100,22 @@ impl YellowstoneGrpc { { match event_pretty { EventPretty::Transaction(transaction_pretty) => { - let trade_raw: TransactionWithStatusMeta = transaction_pretty.tx; - let meta = trade_raw.get_status_meta(); - - if meta.is_none() { - return Ok(()); + let tx = yellowstone_grpc_proto::convert_from::create_tx_with_meta( + transaction_pretty.grpc_tx, + ); + if let Ok(tx) = tx { + let trade_raw: TransactionWithStatusMeta = tx; + let meta = trade_raw.get_status_meta(); + if meta.is_none() { + return Ok(()); + } + let transaction = trade_raw.get_transaction(); + callback(SystemEvent::NewTransfer(TransferInfo { + slot: transaction_pretty.slot, + signature: transaction_pretty.signature.to_string(), + tx: Some(transaction), + })); } - - let transaction = trade_raw.get_transaction(); - - callback(SystemEvent::NewTransfer(TransferInfo { - slot: transaction_pretty.slot, - signature: transaction_pretty.signature.to_string(), - tx: Some(transaction), - })); } _ => {} }