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

perf: optimize event processing and add gRPC instruction parsing

Browse Source
This commit is contained in:
ysq
2025-09-03 15:29:57 +08:00
parent b71a83bf66
commit 7601eb370c
7 changed files with 627 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/streaming/common/event_processor.rs
+27 -46
View File
@@ -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<EventTypeFilter>,
pub(crate) callback: Option<Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>>,
pub(crate) backpressure_config: BackpressureConfig,
/// High-performance lockfree queue for gRPC events
pub(crate) grpc_queue: Arc<SegQueue<(EventPretty, Option<Pubkey>)>>,
/// High-performance lockfree queue for shred events
pub(crate) shred_queue: Arc<SegQueue<(TransactionWithSlot, Option<Pubkey>)>>,
/// Fast O(1) counter for Drop strategy (avoids expensive SegQueue::len())
pub(crate) grpc_pending_count: Arc<AtomicUsize>,
pub(crate) shred_pending_count: Arc<AtomicUsize>,
/// Processing thread control
pub(crate) processing_shutdown: Arc<AtomicBool>,
}
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<dyn Fn(Box<dyn UnifiedEvent>) + 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<Pubkey>,
) -> 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 {
src/streaming/event_parser/common/types.rs
+167
View File
@@ -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<SwapData> {
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<Pubkey> = None;
let mut from_mint: Option<Pubkey> = None;
let mut to_mint: Option<Pubkey> = None;
let mut user_from_token: Option<Pubkey> = None;
let mut user_to_token: Option<Pubkey> = None;
let mut from_vault: Option<Pubkey> = None;
let mut to_vault: Option<Pubkey> = 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
}
}
src/streaming/event_parser/core/macros.rs
+37
View File
@@ -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<prost_types::Timestamp>,
program_received_time_us: i64,
outer_index: i64,
inner_index: Option<i64>,
transaction_index: Option<u64>,
config: &GenericEventParseConfig,
) -> Vec<Box<dyn $crate::streaming::event_parser::core::traits::UnifiedEvent>> {
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<prost_types::Timestamp>,
program_received_time_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
transaction_index: Option<u64>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: std::sync::Arc<
dyn for<'a> Fn(&'a Box<dyn $crate::streaming::event_parser::core::traits::UnifiedEvent>)
+ 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)
}
src/streaming/event_parser/core/traits.rs
+373 -52
View File
@@ -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<Box<dyn UnifiedEvent>>;
/// 从内联指令中解析事件数据
#[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<Timestamp>,
program_received_time_us: i64,
outer_index: i64,
inner_index: Option<i64>,
transaction_index: Option<u64>,
config: &GenericEventParseConfig,
) -> Vec<Box<dyn UnifiedEvent>>;
/// 从指令中解析事件数据
#[allow(clippy::too_many_arguments)]
fn parse_events_from_instruction(
@@ -307,6 +325,80 @@ pub trait EventParser: Send + Sync {
callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + 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<Timestamp>,
program_received_time_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + 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<u64>,
block_time: Option<Timestamp>,
program_received_time_us: i64,
accounts: &[Pubkey],
inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions],
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + 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<u64>,
block_time: Option<Timestamp>,
@@ -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<u64>,
block_time: Option<Timestamp>,
@@ -465,60 +556,86 @@ pub trait EventParser: Send + Sync {
transaction_index: Option<u64>,
callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
) -> anyhow::Result<()> {
let versioned_tx = tx.get_transaction();
let meta = tx.get_status_meta();
let mut address_table_lookups: Vec<Pubkey> = vec![];
let mut inner_instructions: Vec<InnerInstructions> = 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<u8>> = 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<u8>> =
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<Pubkey> = 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<u64>,
block_time: Option<Timestamp>,
program_received_time_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + 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<Timestamp>,
program_received_time_us: i64,
outer_index: i64,
inner_index: Option<i64>,
transaction_index: Option<u64>,
config: &GenericEventParseConfig,
) -> Vec<Box<dyn UnifiedEvent>> {
// 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<Timestamp>,
program_received_time_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
transaction_index: Option<u64>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + 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<Box<dyn UnifiedEvent>> = 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)
}
src/streaming/grpc/pool.rs
+1 -2
View File
@@ -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;
}
}
src/streaming/grpc/types.rs
+7 -7
View File
@@ -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<String, SubscribeRequestFilterTransactions>;
pub type AccountsFilterMap = HashMap<String, SubscribeRequestFilterAccounts>;
#[derive(Clone)]
#[derive(Clone, Debug)]
pub enum EventPretty {
BlockMeta(BlockMetaPretty),
Transaction(TransactionPretty),
@@ -71,8 +74,8 @@ pub struct TransactionPretty {
pub block_time: Option<Timestamp>,
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,
}
}
src/streaming/yellowstone_sub_system.rs
+15 -13
View File
@@ -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),
}));
}
_ => {}
}