From b71a83bf66510f1cce716e02d781f897361db211 Mon Sep 17 00:00:00 2001 From: ysq Date: Tue, 2 Sep 2025 17:27:27 +0800 Subject: [PATCH 1/4] perf: add object pooling and enhanced metrics for streaming optimization - Implement gRPC/shred connection pools for memory efficiency - Add atomic processing time stats with auto-calibration clock - Optimize event parsers and protocol handlers - Refactor metrics system for advanced performance monitoring --- src/streaming/common/config.rs | 2 +- src/streaming/common/event_processor.rs | 6 +- src/streaming/common/metrics.rs | 128 ++--- src/streaming/event_parser/common/mod.rs | 2 +- src/streaming/event_parser/common/types.rs | 20 +- .../event_parser/core/account_event_parser.rs | 12 +- .../event_parser/core/common_event_parser.rs | 17 +- src/streaming/event_parser/core/traits.rs | 112 ++++- .../protocols/block/block_meta_event.rs | 5 +- .../event_parser/protocols/bonk/events.rs | 13 +- .../event_parser/protocols/bonk/types.rs | 12 +- .../event_parser/protocols/pumpfun/events.rs | 8 +- .../event_parser/protocols/pumpfun/types.rs | 8 +- .../event_parser/protocols/pumpswap/events.rs | 10 +- .../event_parser/protocols/pumpswap/types.rs | 8 +- .../protocols/raydium_amm_v4/events.rs | 10 +- .../protocols/raydium_amm_v4/types.rs | 4 +- .../protocols/raydium_clmm/events.rs | 12 +- .../protocols/raydium_clmm/types.rs | 12 +- .../protocols/raydium_cpmm/events.rs | 14 +- .../protocols/raydium_cpmm/types.rs | 8 +- src/streaming/grpc/mod.rs | 2 + src/streaming/grpc/pool.rs | 438 ++++++++++++++++++ src/streaming/grpc/types.rs | 137 +++--- src/streaming/shred/mod.rs | 2 + src/streaming/shred/pool.rs | 156 +++++++ src/streaming/shred/types.rs | 2 +- src/streaming/shred_stream.rs | 7 +- src/streaming/yellowstone_grpc.rs | 11 +- src/streaming/yellowstone_sub_system.rs | 15 +- 30 files changed, 920 insertions(+), 273 deletions(-) create mode 100644 src/streaming/grpc/pool.rs create mode 100644 src/streaming/shred/pool.rs diff --git a/src/streaming/common/config.rs b/src/streaming/common/config.rs index ce21ec8..c314813 100644 --- a/src/streaming/common/config.rs +++ b/src/streaming/common/config.rs @@ -26,7 +26,7 @@ pub struct BackpressureConfig { impl Default for BackpressureConfig { fn default() -> Self { - Self { permits: 1, strategy: BackpressureStrategy::default() } + Self { permits: 3000, strategy: BackpressureStrategy::default() } } } diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 7ae82f7..300dc22 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -1,6 +1,6 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; -use std::time::Instant; + use crossbeam_queue::SegQueue; use solana_sdk::pubkey::Pubkey; @@ -13,7 +13,7 @@ use crate::streaming::common::{ 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::traits::get_high_perf_clock; + use crate::streaming::event_parser::EventParser; use crate::streaming::event_parser::{ core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol, @@ -225,7 +225,7 @@ impl EventProcessor { .unwrap_or_else(|| chrono::Utc::now().timestamp_millis()); let block_meta_event = CommonEventParser::generate_block_meta_event( block_meta_pretty.slot, - &block_meta_pretty.block_hash, + block_meta_pretty.block_hash, block_time_ms, block_meta_pretty.program_received_time_us, ); diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index 402f909..f76e866 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -39,7 +39,8 @@ struct AtomicEventMetrics { events_processed: AtomicU64, events_in_window: AtomicU64, window_start_nanos: AtomicU64, - events_per_second_bits: AtomicU64, // Bit representation of f64 + // Processing time statistics per event type + processing_stats: AtomicProcessingTimeStats, } impl AtomicEventMetrics { @@ -49,7 +50,7 @@ impl AtomicEventMetrics { events_processed: AtomicU64::new(0), events_in_window: AtomicU64::new(0), window_start_nanos: AtomicU64::new(now_nanos), - events_per_second_bits: AtomicU64::new(0), + processing_stats: AtomicProcessingTimeStats::new(), } } @@ -76,18 +77,6 @@ impl AtomicEventMetrics { ) } - /// Atomically update events per second - #[inline] - fn update_events_per_second(&self, eps: f64) { - self.events_per_second_bits.store(eps.to_bits(), Ordering::Relaxed); - } - - /// Get events per second - #[inline] - fn get_events_per_second(&self) -> f64 { - f64::from_bits(self.events_per_second_bits.load(Ordering::Relaxed)) - } - /// Reset window count #[inline] fn reset_window(&self, new_start_nanos: u64) { @@ -99,6 +88,18 @@ impl AtomicEventMetrics { fn get_window_start(&self) -> u64 { self.window_start_nanos.load(Ordering::Relaxed) } + + /// Get processing time statistics for this event type + #[inline] + fn get_processing_stats(&self) -> ProcessingTimeStats { + self.processing_stats.get_stats() + } + + /// Update processing time statistics for this event type + #[inline] + fn update_processing_stats(&self, time_us: f64, event_count: u64) { + self.processing_stats.update(time_us, event_count); + } } /// High-performance atomic processing time statistics @@ -139,7 +140,7 @@ impl AtomicProcessingTimeStats { // Update minimum value, check time difference and reset if over 10 seconds let mut current_min = self.min_time_bits.load(Ordering::Relaxed); let min_timestamp = self.min_time_timestamp_nanos.load(Ordering::Relaxed); - + // Check if min value timestamp exceeds 10 seconds (10_000_000_000 nanoseconds) let min_time_diff_nanos = now_nanos.saturating_sub(min_timestamp); if min_time_diff_nanos > 10_000_000_000 { @@ -148,7 +149,7 @@ impl AtomicProcessingTimeStats { self.min_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); current_min = f64::INFINITY.to_bits(); } - + // If current time is less than min value, update min value and timestamp while time_bits < current_min { match self.min_time_bits.compare_exchange_weak( @@ -236,7 +237,7 @@ pub struct ProcessingTimeStats { pub struct EventMetricsSnapshot { pub process_count: u64, pub events_processed: u64, - pub events_per_second: f64, + pub processing_stats: ProcessingTimeStats, } /// Compatibility structure - complete performance metrics @@ -253,9 +254,12 @@ pub struct PerformanceMetrics { impl PerformanceMetrics { /// Create default performance metrics (compatibility method) pub fn new() -> Self { - let default_metrics = - EventMetricsSnapshot { process_count: 0, events_processed: 0, events_per_second: 0.0 }; let default_stats = ProcessingTimeStats { min_us: 0.0, max_us: 0.0, avg_us: 0.0 }; + let default_metrics = EventMetricsSnapshot { + process_count: 0, + events_processed: 0, + processing_stats: default_stats.clone(), + }; Self { uptime: std::time::Duration::ZERO, @@ -311,9 +315,9 @@ impl HighPerformanceMetrics { pub fn get_event_metrics(&self, event_type: EventType) -> EventMetricsSnapshot { let index = event_type.as_index(); let (process_count, events_processed, _) = self.event_metrics[index].get_counts(); - let events_per_second = self.calculate_real_time_eps(event_type); + let processing_stats = self.event_metrics[index].get_processing_stats(); - EventMetricsSnapshot { process_count, events_processed, events_per_second } + EventMetricsSnapshot { process_count, events_processed, processing_stats } } /// 获取处理时间统计 @@ -328,41 +332,6 @@ impl HighPerformanceMetrics { self.dropped_events_count.load(Ordering::Relaxed) } - /// 计算实时每秒事件数（非阻塞） - fn calculate_real_time_eps(&self, event_type: EventType) -> f64 { - let now_nanos = - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() - as u64; - - let index = event_type.as_index(); - let event_metric = &self.event_metrics[index]; - - let window_start = event_metric.get_window_start(); - let current_window_duration_secs = - (now_nanos.saturating_sub(window_start)) as f64 / 1_000_000_000.0; - let events_in_window = event_metric.events_in_window.load(Ordering::Relaxed); - - // 优先级1: 当前窗口实时数据（≥2秒且有事件） - if current_window_duration_secs >= 2.0 && events_in_window > 0 { - return events_in_window as f64 / current_window_duration_secs; - } - - // 优先级2: 上一个窗口的结果 - let stored_eps = event_metric.get_events_per_second(); - if stored_eps > 0.0 { - return stored_eps; - } - - // 优先级3: 总体平均值（≥3秒运行时间） - let total_duration_secs = self.get_uptime_seconds(); - let total_events = event_metric.events_processed.load(Ordering::Relaxed); - if total_duration_secs >= 3.0 && total_events > 0 { - return total_events as f64 / total_duration_secs; - } - - 0.0 - } - /// 更新窗口指标（后台任务调用） fn update_window_metrics(&self, event_type: EventType, window_duration_nanos: u64) { let now_nanos = @@ -374,14 +343,6 @@ impl HighPerformanceMetrics { let window_start = event_metric.get_window_start(); if now_nanos.saturating_sub(window_start) >= window_duration_nanos { - let events_in_window = event_metric.events_in_window.load(Ordering::Relaxed); - let window_duration_secs = window_duration_nanos as f64 / 1_000_000_000.0; - - if window_duration_secs > 0.001 && events_in_window > 0 { - let eps = events_in_window as f64 / window_duration_secs; - event_metric.update_events_per_second(eps); - } - event_metric.reset_window(now_nanos); } } @@ -455,10 +416,15 @@ impl MetricsManager { return; } - // 原子更新事件计数 - self.metrics.event_metrics[event_type.as_index()].add_events_processed(count); + let index = event_type.as_index(); - // 原子更新处理时间统计 + // 原子更新事件计数 + self.metrics.event_metrics[index].add_events_processed(count); + + // 原子更新该事件类型的处理时间统计 + self.metrics.event_metrics[index].update_processing_stats(processing_time_us, count); + + // 保持全局处理时间统计的兼容性 self.metrics.processing_stats.update(processing_time_us, count); } @@ -506,35 +472,25 @@ impl MetricsManager { println!("\n⚠️ Dropped Events: {}", dropped_count); } - // 打印事件指标表格 - println!("┌─────────────┬──────────────┬──────────────────┬─────────────────┐"); - println!("│ Event Type │ Process Count│ Events Processed │ Events/Second │"); - println!("├─────────────┼──────────────┼──────────────────┼─────────────────┤"); + // 打印事件指标表格（包含处理时间统计） + println!("┌─────────────┬──────────────┬──────────────────┬─────────────┬─────────────┬─────────────┐"); + println!("│ Event Type │ Process Count│ Events Processed │ Avg Time(μs)│ Min 10s(μs) │ Max 10s(μs) │"); + println!("├─────────────┼──────────────┼──────────────────┼─────────────┼─────────────┼─────────────┤"); for event_type in [EventType::Transaction, EventType::Account, EventType::BlockMeta] { let metrics = self.get_event_metrics(event_type); println!( - "│ {:11} │ {:12} │ {:16} │ {:13.2} │", + "│ {:11} │ {:12} │ {:16} │ {:9.2} │ {:9.2} │ {:9.2} │", event_type.name(), metrics.process_count, metrics.events_processed, - metrics.events_per_second + metrics.processing_stats.avg_us, + metrics.processing_stats.min_us, + metrics.processing_stats.max_us ); } - println!("└─────────────┴──────────────┴──────────────────┴─────────────────┘"); - - // 打印处理时间统计表格 - let stats = self.get_processing_stats(); - println!("\n⏱️ Processing Time Statistics"); - println!("┌───────────────────────┬─────────────┐"); - println!("│ Metric │ Value (us) │"); - println!("├───────────────────────┼─────────────┤"); - println!("│ Average │ {:9.2} │", stats.avg_us); - println!("│ Minimum within 10s │ {:9.2} │", stats.min_us); - println!("│ Maximum within 10s │ {:9.2} │", stats.max_us); - println!("└───────────────────────┴─────────────┘"); - + println!("└─────────────┴──────────────┴──────────────────┴─────────────┴─────────────┴─────────────┘"); println!(); } diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index 3e9d088..6784e91 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -12,7 +12,7 @@ macro_rules! impl_unified_event { self.metadata.event_type.clone() } - fn signature(&self) -> &str { + fn signature(&self) -> &solana_sdk::signature::Signature { &self.metadata.signature } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index b086d8f..2daa630 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -1,7 +1,7 @@ use borsh::{BorshDeserialize, BorshSerialize}; use crossbeam_queue::ArrayQueue; use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; +use solana_sdk::{pubkey::Pubkey, signature::Signature}; use std::{borrow::Cow, fmt, str::FromStr, sync::Arc}; use crate::{ @@ -282,15 +282,13 @@ pub struct SwapData { pub to_mint: Pubkey, pub from_amount: u64, pub to_amount: u64, - pub description: Option, + pub description: Option>, } /// Event metadata -#[derive( - Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, -)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct EventMetadata { - pub signature: Cow<'static, str>, + pub signature: Signature, pub slot: u64, pub transaction_index: Option, // 新增：交易在slot中的索引 pub block_time: i64, @@ -308,7 +306,7 @@ pub struct EventMetadata { impl EventMetadata { #[allow(clippy::too_many_arguments)] pub fn new( - signature: Cow<'static, str>, + signature: Signature, slot: u64, block_time: i64, block_time_ms: i64, @@ -413,7 +411,7 @@ pub fn parse_swap_data_from_next_instructions( }, RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { user = Some(e.payer); - swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".to_string()); + 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); @@ -430,7 +428,7 @@ pub fn parse_swap_data_from_next_instructions( }, RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { user = Some(e.user_source_owner); - swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".to_string()); + 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); @@ -453,12 +451,12 @@ pub fn parse_swap_data_from_next_instructions( 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 => { diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 0efa7be..001b263 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -1,4 +1,3 @@ -use std::borrow::Cow; use std::collections::HashMap; use std::sync::OnceLock; @@ -7,7 +6,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::common::SimdUtils; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; -use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock}; +use crate::streaming::event_parser::core::traits::{elapsed_micros_since, UnifiedEvent}; use crate::streaming::event_parser::protocols::bonk::parser::BONK_PROGRAM_ID; use crate::streaming::event_parser::protocols::pumpfun::parser::PUMPFUN_PROGRAM_ID; use crate::streaming::event_parser::protocols::pumpswap::parser::PUMPSWAP_PROGRAM_ID; @@ -177,12 +176,11 @@ impl AccountEventParser { if account.owner == config.program_id && SimdUtils::fast_discriminator_match(&account.data, config.account_discriminator) { - let signature_str = Cow::Owned(account.signature.to_string()); let event = (config.account_parser)( &account, EventMetadata { slot: account.slot, - signature: signature_str, + signature: account.signature, protocol: config.protocol_type, event_type: config.event_type, program_id: config.program_id, @@ -191,9 +189,9 @@ impl AccountEventParser { }, ); if let Some(mut event) = event { - event.set_program_handle_time_consuming_us( - get_high_perf_clock().elapsed_micros_since(account.program_received_time_us), - ); + event.set_program_handle_time_consuming_us(elapsed_micros_since( + account.program_received_time_us, + )); return Some(event); } } diff --git a/src/streaming/event_parser/core/common_event_parser.rs b/src/streaming/event_parser/core/common_event_parser.rs index ef44d96..822d1e8 100644 --- a/src/streaming/event_parser/core/common_event_parser.rs +++ b/src/streaming/event_parser/core/common_event_parser.rs @@ -1,4 +1,4 @@ -use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock}; +use crate::streaming::event_parser::core::traits::{elapsed_micros_since, UnifiedEvent}; use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent; pub struct CommonEventParser {} @@ -6,19 +6,14 @@ pub struct CommonEventParser {} impl CommonEventParser { pub fn generate_block_meta_event( slot: u64, - block_hash: &str, + block_hash: String, block_time_ms: i64, program_received_time_us: i64, ) -> Box { - let mut block_meta_event = BlockMetaEvent::new( - slot, - block_hash.to_string(), - block_time_ms, - program_received_time_us, - ); - block_meta_event.set_program_handle_time_consuming_us( - get_high_perf_clock().elapsed_micros_since(program_received_time_us), - ); + let mut block_meta_event = + BlockMetaEvent::new(slot, block_hash, block_time_ms, program_received_time_us); + block_meta_event + .set_program_handle_time_consuming_us(elapsed_micros_since(program_received_time_us)); Box::new(block_meta_event) } } diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 65c93de..d45e2dd 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -29,22 +29,53 @@ use crate::streaming::event_parser::{ }, }; -/// 高性能时钟管理器，减少系统调用开销 +/// 高性能时钟管理器，减少系统调用开销并最小化延迟 #[derive(Debug)] pub struct HighPerformanceClock { /// 基准时间点（程序启动时的单调时钟时间） base_instant: Instant, /// 基准时间点对应的UTC时间戳（微秒） base_timestamp_us: i64, + /// 上次校准时间（用于检测是否需要重新校准） + last_calibration: Instant, + /// 校准间隔（秒） + calibration_interval_secs: u64, } impl HighPerformanceClock { /// 创建新的高性能时钟 pub fn new() -> Self { - let base_instant = Instant::now(); - let base_timestamp_us = chrono::Utc::now().timestamp_micros(); + Self::new_with_calibration_interval(300) // 默认5分钟校准一次 + } - Self { base_instant, base_timestamp_us } + /// 创建带自定义校准间隔的高性能时钟 + pub fn new_with_calibration_interval(calibration_interval_secs: u64) -> Self { + // 通过多次采样来减少初始化误差 + let mut best_offset = i64::MAX; + let mut best_instant = Instant::now(); + let mut best_timestamp = chrono::Utc::now().timestamp_micros(); + + // 进行3次采样，选择延迟最小的 + for _ in 0..3 { + let instant_before = Instant::now(); + let timestamp = chrono::Utc::now().timestamp_micros(); + let instant_after = Instant::now(); + + let sample_latency = instant_after.duration_since(instant_before).as_nanos() as i64; + + if sample_latency < best_offset { + best_offset = sample_latency; + best_instant = instant_before; + best_timestamp = timestamp; + } + } + + Self { + base_instant: best_instant, + base_timestamp_us: best_timestamp, + last_calibration: best_instant, + calibration_interval_secs, + } } /// 获取当前时间戳（微秒），使用单调时钟计算，避免系统调用 @@ -54,11 +85,53 @@ impl HighPerformanceClock { self.base_timestamp_us + elapsed.as_micros() as i64 } + /// 获取高精度当前时间戳（微秒），在必要时进行校准 + pub fn now_micros_with_calibration(&mut self) -> i64 { + // 检查是否需要重新校准 + if self.last_calibration.elapsed().as_secs() >= self.calibration_interval_secs { + self.recalibrate(); + } + self.now_micros() + } + + /// 重新校准时钟，减少累积漂移 + fn recalibrate(&mut self) { + let current_monotonic = Instant::now(); + let current_utc = chrono::Utc::now().timestamp_micros(); + + // 计算预期的UTC时间戳（基于单调时钟） + let expected_utc = self.base_timestamp_us + + current_monotonic.duration_since(self.base_instant).as_micros() as i64; + + // 计算漂移量 + let drift_us = current_utc - expected_utc; + + // 如果漂移超过1毫秒，进行校准 + if drift_us.abs() > 1000 { + self.base_instant = current_monotonic; + self.base_timestamp_us = current_utc; + } + + self.last_calibration = current_monotonic; + } + /// 计算从指定时间戳到现在的消耗时间（微秒） #[inline(always)] pub fn elapsed_micros_since(&self, start_timestamp_us: i64) -> i64 { self.now_micros() - start_timestamp_us } + + /// 获取高精度纳秒时间戳 + #[inline(always)] + pub fn now_nanos(&self) -> i128 { + let elapsed = self.base_instant.elapsed(); + (self.base_timestamp_us as i128 * 1000) + elapsed.as_nanos() as i128 + } + + /// 重置时钟（强制重新初始化） + pub fn reset(&mut self) { + *self = Self::new_with_calibration_interval(self.calibration_interval_secs); + } } impl Default for HighPerformanceClock { @@ -67,14 +140,21 @@ impl Default for HighPerformanceClock { } } -/// 全局高性能时钟实例（使用OnceCell避免重复初始化） +/// 全局高性能时钟实例 static HIGH_PERF_CLOCK: once_cell::sync::OnceCell = once_cell::sync::OnceCell::new(); -/// 获取全局高性能时钟实例 +/// 获取全局高性能时钟实例（最简单的实现） #[inline(always)] -pub fn get_high_perf_clock() -> &'static HighPerformanceClock { - HIGH_PERF_CLOCK.get_or_init(HighPerformanceClock::new) +pub fn get_high_perf_clock() -> i64 { + let clock = HIGH_PERF_CLOCK.get_or_init(HighPerformanceClock::new); + clock.now_micros() +} + +/// 计算从指定时间戳到现在的消耗时间（微秒） +#[inline(always)] +pub fn elapsed_micros_since(start_timestamp_us: i64) -> i64 { + get_high_perf_clock() - start_timestamp_us } /// 轻量级事件包装器，避免频繁的Box分配 @@ -147,7 +227,7 @@ pub trait UnifiedEvent: Debug + Send + Sync { fn event_type(&self) -> EventType; /// Get transaction signature - fn signature(&self) -> &str; + fn signature(&self) -> &Signature; /// Get slot number fn slot(&self) -> u64; @@ -535,7 +615,7 @@ pub trait EventParser: Send + Sync { let slot = transaction.slot; let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 }); - let program_received_time_us = chrono::Utc::now().timestamp_micros(); + let program_received_time_us = get_high_perf_clock(); let bot_wallet = None; let transaction_index = None; // 解析指令事件 @@ -714,11 +794,10 @@ impl GenericEventParser { transaction_index: Option, ) -> Option> { if let Some(parser) = config.inner_instruction_parser { - let signature_str = Cow::Owned(signature.to_string()); let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( - signature_str, + signature, slot, timestamp.seconds, block_time_ms, @@ -752,11 +831,10 @@ impl GenericEventParser { transaction_index: Option, ) -> Option> { if let Some(parser) = config.instruction_parser { - let signature_str = Cow::Owned(signature.to_string()); let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( - signature_str, + signature, slot, timestamp.seconds, block_time_ms, @@ -941,9 +1019,9 @@ impl EventParser for GenericEventParser { event.merge(&*inner_instruction_event); } // 设置处理时间（使用高性能时钟） - event.set_program_handle_time_consuming_us( - get_high_perf_clock().elapsed_micros_since(program_received_time_us), - ); + event.set_program_handle_time_consuming_us(elapsed_micros_since( + program_received_time_us, + )); event = process_event(event, bot_wallet); callback(&event); } diff --git a/src/streaming/event_parser/protocols/block/block_meta_event.rs b/src/streaming/event_parser/protocols/block/block_meta_event.rs index 8fe4051..392e24b 100644 --- a/src/streaming/event_parser/protocols/block/block_meta_event.rs +++ b/src/streaming/event_parser/protocols/block/block_meta_event.rs @@ -1,9 +1,8 @@ -use std::borrow::Cow; - use crate::impl_unified_event; use crate::streaming::event_parser::common::{types::EventType, EventMetadata}; use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; +use solana_sdk::signature::Signature; /// Block元数据事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] @@ -22,7 +21,7 @@ impl BlockMetaEvent { program_received_time_us: i64, ) -> Self { let metadata = EventMetadata::new( - Cow::Borrowed(""), + Signature::default(), slot, block_time_ms / 1000, block_time_ms, diff --git a/src/streaming/event_parser/protocols/bonk/events.rs b/src/streaming/event_parser/protocols/bonk/events.rs index cb450ae..6571664 100755 --- a/src/streaming/event_parser/protocols/bonk/events.rs +++ b/src/streaming/event_parser/protocols/bonk/events.rs @@ -237,6 +237,7 @@ impl_unified_event!( // Migrate to CP Swap event #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct BonkMigrateToCpswapEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub payer: Pubkey, pub base_mint: Pubkey, @@ -276,10 +277,10 @@ impl_unified_event!(BonkMigrateToCpswapEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct BonkPoolStateAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool_state: PoolState, } @@ -289,10 +290,10 @@ impl_unified_event!(BonkPoolStateAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct BonkGlobalConfigAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub global_config: GlobalConfig, } @@ -302,10 +303,10 @@ impl_unified_event!(BonkGlobalConfigAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct BonkPlatformConfigAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub platform_config: PlatformConfig, } diff --git a/src/streaming/event_parser/protocols/bonk/types.rs b/src/streaming/event_parser/protocols/bonk/types.rs index 476f3ae..2e82b12 100755 --- a/src/streaming/event_parser/protocols/bonk/types.rs +++ b/src/streaming/event_parser/protocols/bonk/types.rs @@ -142,10 +142,10 @@ pub fn pool_state_parser( if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { Some(Box::new(BonkPoolStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool_state, })) @@ -193,10 +193,10 @@ pub fn global_config_parser( if let Some(global_config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { Some(Box::new(BonkGlobalConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, global_config, })) @@ -241,10 +241,10 @@ pub fn platform_config_parser( { Some(Box::new(BonkPlatformConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, platform_config, })) diff --git a/src/streaming/event_parser/protocols/pumpfun/events.rs b/src/streaming/event_parser/protocols/pumpfun/events.rs index 5e06a84..32212e2 100755 --- a/src/streaming/event_parser/protocols/pumpfun/events.rs +++ b/src/streaming/event_parser/protocols/pumpfun/events.rs @@ -228,10 +228,10 @@ impl_unified_event!( pub struct PumpFunBondingCurveAccountEvent { #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub bonding_curve: BondingCurve, } @@ -243,10 +243,10 @@ impl_unified_event!(PumpFunBondingCurveAccountEvent,); pub struct PumpFunGlobalAccountEvent { #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub global: Global, } diff --git a/src/streaming/event_parser/protocols/pumpfun/types.rs b/src/streaming/event_parser/protocols/pumpfun/types.rs index c2ffdf1..d419996 100644 --- a/src/streaming/event_parser/protocols/pumpfun/types.rs +++ b/src/streaming/event_parser/protocols/pumpfun/types.rs @@ -41,10 +41,10 @@ pub fn bonding_curve_parser( if let Some(bonding_curve) = bonding_curve_decode(&account.data[8..BONDING_CURVE_SIZE + 8]) { Some(Box::new(PumpFunBondingCurveAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, bonding_curve, })) @@ -91,10 +91,10 @@ pub fn global_parser( if let Some(global) = global_decode(&account.data[8..GLOBAL_SIZE + 8]) { Some(Box::new(PumpFunGlobalAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, global, })) diff --git a/src/streaming/event_parser/protocols/pumpswap/events.rs b/src/streaming/event_parser/protocols/pumpswap/events.rs index 63a85b7..c3d2322 100755 --- a/src/streaming/event_parser/protocols/pumpswap/events.rs +++ b/src/streaming/event_parser/protocols/pumpswap/events.rs @@ -366,11 +366,12 @@ impl_unified_event!( /// 全局配置 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapGlobalConfigAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub global_config: GlobalConfig, } @@ -379,11 +380,12 @@ impl_unified_event!(PumpSwapGlobalConfigAccountEvent,); /// 池 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpSwapPoolAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool: Pool, } diff --git a/src/streaming/event_parser/protocols/pumpswap/types.rs b/src/streaming/event_parser/protocols/pumpswap/types.rs index b58c0dc..a9d917b 100644 --- a/src/streaming/event_parser/protocols/pumpswap/types.rs +++ b/src/streaming/event_parser/protocols/pumpswap/types.rs @@ -43,10 +43,10 @@ pub fn global_config_parser( if let Some(config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { Some(Box::new(PumpSwapGlobalConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, global_config: config, })) @@ -88,10 +88,10 @@ pub fn pool_parser( if let Some(pool) = pool_decode(&account.data[8..POOL_SIZE + 8]) { Some(Box::new(PumpSwapPoolAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool: pool, })) diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs index cd4eb19..9f82f6f 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs @@ -9,6 +9,7 @@ use solana_sdk::pubkey::Pubkey; /// 交易 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4SwapEvent { + #[borsh(skip)] pub metadata: EventMetadata, // base in pub amount_in: u64, @@ -42,6 +43,7 @@ impl_unified_event!(RaydiumAmmV4SwapEvent,); /// 添加流动性 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4DepositEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub max_coin_amount: u64, pub max_pc_amount: u64, @@ -67,6 +69,7 @@ impl_unified_event!(RaydiumAmmV4DepositEvent,); /// 初始化 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4Initialize2Event { + #[borsh(skip)] pub metadata: EventMetadata, pub nonce: u8, pub open_time: u64, @@ -100,6 +103,7 @@ impl_unified_event!(RaydiumAmmV4Initialize2Event,); /// 移除流动性 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4WithdrawEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub amount: u64, @@ -131,6 +135,7 @@ impl_unified_event!(RaydiumAmmV4WithdrawEvent,); /// 提现 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4WithdrawPnlEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub token_program: Pubkey, @@ -156,11 +161,12 @@ impl_unified_event!(RaydiumAmmV4WithdrawPnlEvent,); /// 池信息 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumAmmV4AmmInfoAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub amm_info: AmmInfo, } diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs index 63fac73..e2f3078 100644 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs @@ -96,10 +96,10 @@ pub fn amm_info_parser( if let Some(amm_info) = amm_info_decode(&account.data[..AMM_INFO_SIZE]) { Some(Box::new(RaydiumAmmV4AmmInfoAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, amm_info: amm_info, })) diff --git a/src/streaming/event_parser/protocols/raydium_clmm/events.rs b/src/streaming/event_parser/protocols/raydium_clmm/events.rs index f8834fa..dd405cf 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/events.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/events.rs @@ -223,10 +223,10 @@ impl_unified_event!(RaydiumClmmOpenPositionV2Event,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RaydiumClmmAmmConfigAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub amm_config: AmmConfig, } @@ -236,10 +236,10 @@ impl_unified_event!(RaydiumClmmAmmConfigAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RaydiumClmmPoolStateAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool_state: PoolState, } @@ -249,10 +249,10 @@ impl_unified_event!(RaydiumClmmPoolStateAccountEvent,); #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RaydiumClmmTickArrayStateAccountEvent { pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub tick_array_state: TickArrayState, } diff --git a/src/streaming/event_parser/protocols/raydium_clmm/types.rs b/src/streaming/event_parser/protocols/raydium_clmm/types.rs index 06cc524..c8a8803 100644 --- a/src/streaming/event_parser/protocols/raydium_clmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/types.rs @@ -47,10 +47,10 @@ pub fn amm_config_parser( if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { Some(Box::new(RaydiumClmmAmmConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, amm_config: amm_config, })) @@ -135,10 +135,10 @@ pub fn pool_state_parser( if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { Some(Box::new(RaydiumClmmPoolStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool_state: pool_state, })) @@ -218,10 +218,10 @@ pub fn tick_array_state_parser( { Some(Box::new(RaydiumClmmTickArrayStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, tick_array_state: tick_array_state, })) diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs b/src/streaming/event_parser/protocols/raydium_cpmm/events.rs index 6260e48..d8c1b7a 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/events.rs @@ -10,6 +10,7 @@ use solana_sdk::pubkey::Pubkey; /// 交易 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmSwapEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub amount_in: u64, pub minimum_amount_out: u64, @@ -35,6 +36,7 @@ impl_unified_event!(RaydiumCpmmSwapEvent,); /// 存款 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmDepositEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub lp_token_amount: u64, pub maximum_token0_amount: u64, @@ -59,6 +61,7 @@ impl_unified_event!(RaydiumCpmmDepositEvent,); /// 初始化 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmInitializeEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub init_amount0: u64, pub init_amount1: u64, @@ -90,6 +93,7 @@ impl_unified_event!(RaydiumCpmmInitializeEvent,); /// 提款 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmWithdrawEvent { + #[borsh(skip)] pub metadata: EventMetadata, pub lp_token_amount: u64, pub minimum_token0_amount: u64, @@ -115,11 +119,12 @@ impl_unified_event!(RaydiumCpmmWithdrawEvent,); /// 池配置 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmAmmConfigAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub amm_config: AmmConfig, } @@ -128,11 +133,12 @@ impl_unified_event!(RaydiumCpmmAmmConfigAccountEvent,); /// 池状态 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct RaydiumCpmmPoolStateAccountEvent { + #[borsh(skip)] pub metadata: EventMetadata, - pub pubkey: String, + pub pubkey: Pubkey, pub executable: bool, pub lamports: u64, - pub owner: String, + pub owner: Pubkey, pub rent_epoch: u64, pub pool_state: PoolState, } diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs index 29555c7..532690f 100644 --- a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs @@ -46,10 +46,10 @@ pub fn amm_config_parser( if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { Some(Box::new(RaydiumCpmmAmmConfigAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, amm_config: amm_config, })) @@ -104,10 +104,10 @@ pub fn pool_state_parser( if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { Some(Box::new(RaydiumCpmmPoolStateAccountEvent { metadata, - pubkey: account.pubkey.to_string(), + pubkey: account.pubkey, executable: account.executable, lamports: account.lamports, - owner: account.owner.to_string(), + owner: account.owner, rent_epoch: account.rent_epoch, pool_state: pool_state, })) diff --git a/src/streaming/grpc/mod.rs b/src/streaming/grpc/mod.rs index 0d91713..6618f4e 100644 --- a/src/streaming/grpc/mod.rs +++ b/src/streaming/grpc/mod.rs @@ -1,10 +1,12 @@ // gRPC 相关模块 pub mod connection; +pub mod pool; pub mod subscription; pub mod types; // 重新导出主要类型 pub use connection::*; +pub use pool::*; pub use subscription::*; pub use types::*; diff --git a/src/streaming/grpc/pool.rs b/src/streaming/grpc/pool.rs new file mode 100644 index 0000000..5371a54 --- /dev/null +++ b/src/streaming/grpc/pool.rs @@ -0,0 +1,438 @@ +use solana_sdk::{pubkey::Pubkey, signature::Signature}; +use std::collections::VecDeque; +use std::ops::DerefMut; +use std::sync::{Arc, Mutex}; +use yellowstone_grpc_proto::{ + geyser::{SubscribeUpdateAccount, SubscribeUpdateBlockMeta, SubscribeUpdateTransaction}, + prost_types::Timestamp, +}; + +use super::types::{AccountPretty, BlockMetaPretty, TransactionPretty}; +use crate::streaming::event_parser::core::traits::get_high_perf_clock; + +/// 通用对象池特征 +pub trait ObjectPool { + fn acquire(&self) -> PooledObject; + fn return_object(&self, obj: Box); +} + +/// 带自动归还的智能指针 +pub struct PooledObject { + object: Option>, + pool: Arc>>>, + max_size: usize, +} + +impl PooledObject { + fn new(object: Box, pool: Arc>>>, max_size: usize) -> Self { + Self { object: Some(object), pool, max_size } + } +} + +impl Drop for PooledObject { + fn drop(&mut self) { + if let Some(obj) = self.object.take() { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + pool.push_back(obj); + } + // 超过最大容量时直接丢弃 + } + } +} + +impl std::ops::Deref for PooledObject { + type Target = T; + + fn deref(&self) -> &Self::Target { + self.object.as_ref().unwrap() + } +} + +impl std::ops::DerefMut for PooledObject { + fn deref_mut(&mut self) -> &mut Self::Target { + self.object.as_mut().unwrap() + } +} + +/// AccountPretty 对象池 +pub struct AccountPrettyPool { + pool: Arc>>>, + max_size: usize, +} + +impl AccountPrettyPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(AccountPretty::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledAccountPretty { + let mut pool = self.pool.lock().unwrap(); + let account = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(AccountPretty::default()), + }; + + PooledAccountPretty { account, pool: Arc::clone(&self.pool), max_size: self.max_size } + } +} + +/// 带自动归还的 AccountPretty +pub struct PooledAccountPretty { + account: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledAccountPretty { + /// 从 gRPC 更新重置数据 + pub fn reset_from_update(&mut self, account_update: SubscribeUpdateAccount) { + let account_info = account_update.account.unwrap(); + + self.account.slot = account_update.slot; + self.account.signature = if let Some(txn_signature) = account_info.txn_signature { + Signature::try_from(txn_signature.as_slice()).expect("valid signature") + } else { + Signature::default() + }; + self.account.pubkey = + Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"); + self.account.executable = account_info.executable; + self.account.lamports = account_info.lamports; + self.account.owner = Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"); + self.account.rent_epoch = account_info.rent_epoch; + + // 优化数据字段的重用 + let new_data = account_info.data; + if self.account.data.capacity() >= new_data.len() { + self.account.data.clear(); + self.account.data.extend_from_slice(&new_data); + } else { + self.account.data = new_data; + } + + self.account.program_received_time_us = get_high_perf_clock(); + } +} + +impl Drop for PooledAccountPretty { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理敏感数据 + self.account.data.clear(); + self.account.signature = Signature::default(); + self.account.pubkey = Pubkey::default(); + self.account.owner = Pubkey::default(); + pool.push_back(std::mem::take(&mut self.account)); + } + } +} + +impl std::ops::Deref for PooledAccountPretty { + type Target = AccountPretty; + + fn deref(&self) -> &Self::Target { + &self.account + } +} + +impl std::ops::DerefMut for PooledAccountPretty { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.account + } +} + +/// BlockMetaPretty 对象池 +pub struct BlockMetaPrettyPool { + pool: Arc>>>, + max_size: usize, +} + +impl BlockMetaPrettyPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(BlockMetaPretty::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledBlockMetaPretty { + let mut pool = self.pool.lock().unwrap(); + let block_meta = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(BlockMetaPretty::default()), + }; + + PooledBlockMetaPretty { block_meta, pool: Arc::clone(&self.pool), max_size: self.max_size } + } +} + +/// 带自动归还的 BlockMetaPretty +pub struct PooledBlockMetaPretty { + block_meta: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledBlockMetaPretty { + /// 从 gRPC 更新重置数据 + pub fn reset_from_update( + &mut self, + block_update: SubscribeUpdateBlockMeta, + block_time: Option, + ) { + self.block_meta.slot = block_update.slot; + self.block_meta.block_hash = block_update.blockhash; + self.block_meta.block_time = block_time; + self.block_meta.program_received_time_us = get_high_perf_clock(); + } +} + +impl Drop for PooledBlockMetaPretty { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理数据 + self.block_meta.block_hash.clear(); + self.block_meta.block_time = None; + pool.push_back(std::mem::take(&mut self.block_meta)); + } + } +} + +impl std::ops::Deref for PooledBlockMetaPretty { + type Target = BlockMetaPretty; + + fn deref(&self) -> &Self::Target { + &self.block_meta + } +} + +impl std::ops::DerefMut for PooledBlockMetaPretty { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.block_meta + } +} + +/// TransactionPretty 对象池 +pub struct TransactionPrettyPool { + pool: Arc>>>, + max_size: usize, +} + +impl TransactionPrettyPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(TransactionPretty::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledTransactionPretty { + let mut pool = self.pool.lock().unwrap(); + let transaction = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(TransactionPretty::default()), + }; + + PooledTransactionPretty { + transaction, + pool: Arc::clone(&self.pool), + max_size: self.max_size, + } + } +} + +/// 带自动归还的 TransactionPretty +pub struct PooledTransactionPretty { + transaction: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledTransactionPretty { + /// 从 gRPC 更新重置数据 + pub fn reset_from_update( + &mut self, + tx_update: SubscribeUpdateTransaction, + block_time: Option, + ) { + let tx = tx_update.transaction.expect("should be defined"); + + self.transaction.slot = tx_update.slot; + self.transaction.transaction_index = Some(tx.index); + self.transaction.block_time = block_time; + self.transaction.block_hash.clear(); // 重置 block_hash + 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(); + } +} + +impl Drop for PooledTransactionPretty { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理数据 + self.transaction.block_hash.clear(); + self.transaction.block_time = None; + self.transaction.signature = Signature::default(); + pool.push_back(std::mem::take(&mut self.transaction)); + } + } +} + +impl std::ops::Deref for PooledTransactionPretty { + type Target = TransactionPretty; + + fn deref(&self) -> &Self::Target { + &self.transaction + } +} + +impl std::ops::DerefMut for PooledTransactionPretty { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.transaction + } +} + +/// EventPretty 对象池（组合池） +pub struct EventPrettyPool { + account_pool: AccountPrettyPool, + block_pool: BlockMetaPrettyPool, + transaction_pool: TransactionPrettyPool, +} + +impl EventPrettyPool { + pub fn new() -> Self { + Self { + account_pool: AccountPrettyPool::new(10000, 20000), + block_pool: BlockMetaPrettyPool::new(500, 1000), + transaction_pool: TransactionPrettyPool::new(10000, 20000), + } + } + + /// 获取账户事件对象 + pub fn acquire_account(&self) -> PooledAccountPretty { + self.account_pool.acquire() + } + + /// 获取区块事件对象 + pub fn acquire_block(&self) -> PooledBlockMetaPretty { + self.block_pool.acquire() + } + + /// 获取交易事件对象 + pub fn acquire_transaction(&self) -> PooledTransactionPretty { + self.transaction_pool.acquire() + } +} + +/// 对象池管理器（单例） +pub struct PoolManager { + event_pool: EventPrettyPool, +} + +impl PoolManager { + pub fn new() -> Self { + Self { event_pool: EventPrettyPool::new() } + } + + pub fn get_event_pool(&self) -> &EventPrettyPool { + &self.event_pool + } +} + +impl Default for PoolManager { + fn default() -> Self { + Self::new() + } +} + +/// 工厂函数用于创建优化的 EventPretty +impl EventPrettyPool { + /// 创建账户事件 - 使用对象池优化 + pub fn create_account_event_optimized(&self, update: SubscribeUpdateAccount) -> AccountPretty { + let mut pooled_account = self.acquire_account(); + pooled_account.reset_from_update(update); + // 移动数据而不是克隆，避免多余的内存分配 + let result = std::mem::replace(pooled_account.deref_mut(), AccountPretty::default()); + result + } + + /// 创建区块事件 - 使用对象池优化 + pub fn create_block_event_optimized( + &self, + update: SubscribeUpdateBlockMeta, + block_time: Option, + ) -> BlockMetaPretty { + let mut pooled_block = self.acquire_block(); + pooled_block.reset_from_update(update, block_time); + // 移动数据而不是克隆 + let result = std::mem::replace(pooled_block.deref_mut(), BlockMetaPretty::default()); + result + } + + /// 创建交易事件 - 使用对象池优化 + pub fn create_transaction_event_optimized( + &self, + update: SubscribeUpdateTransaction, + block_time: Option, + ) -> TransactionPretty { + let mut pooled_tx = self.acquire_transaction(); + pooled_tx.reset_from_update(update, block_time); + // 移动数据而不是克隆 + let result = std::mem::replace(pooled_tx.deref_mut(), TransactionPretty::default()); + result + } +} + +// 全局池管理器实例 +lazy_static::lazy_static! { + pub static ref GLOBAL_POOL_MANAGER: PoolManager = PoolManager::new(); +} + +/// 便捷的全局工厂函数 +pub mod factory { + use super::*; + + /// 使用对象池创建账户事件（推荐用于高性能场景） + pub fn create_account_pretty_pooled(update: SubscribeUpdateAccount) -> AccountPretty { + GLOBAL_POOL_MANAGER.get_event_pool().create_account_event_optimized(update) + } + + /// 使用对象池创建区块事件（推荐用于高性能场景） + pub fn create_block_meta_pretty_pooled( + update: SubscribeUpdateBlockMeta, + block_time: Option, + ) -> BlockMetaPretty { + GLOBAL_POOL_MANAGER.get_event_pool().create_block_event_optimized(update, block_time) + } + + /// 使用对象池创建交易事件（推荐用于高性能场景） + pub fn create_transaction_pretty_pooled( + update: SubscribeUpdateTransaction, + block_time: Option, + ) -> TransactionPretty { + GLOBAL_POOL_MANAGER.get_event_pool().create_transaction_event_optimized(update, block_time) + } +} diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index 7cfd946..1c450de 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -1,11 +1,8 @@ use solana_sdk::{pubkey::Pubkey, signature::Signature}; -use solana_transaction_status::TransactionWithStatusMeta; +use solana_transaction_status::{TransactionWithStatusMeta, VersionedTransactionWithStatusMeta}; use std::{collections::HashMap, fmt}; use yellowstone_grpc_proto::{ - geyser::{ - SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions, SubscribeUpdateAccount, - SubscribeUpdateBlockMeta, SubscribeUpdateTransaction, - }, + geyser::{SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions}, prost_types::Timestamp, }; @@ -19,7 +16,7 @@ pub enum EventPretty { Account(AccountPretty), } -#[derive(Clone)] +#[derive(Clone, Default)] pub struct AccountPretty { pub slot: u64, pub signature: Signature, @@ -47,7 +44,7 @@ impl fmt::Debug for AccountPretty { } } -#[derive(Clone)] +#[derive(Clone, Default)] pub struct BlockMetaPretty { pub slot: u64, pub block_hash: String, @@ -90,63 +87,81 @@ impl fmt::Debug for TransactionPretty { } } -impl From for AccountPretty { - fn from(account: SubscribeUpdateAccount) -> Self { - let account_info = account.account.unwrap(); +impl Default for TransactionPretty { + fn default() -> Self { Self { - slot: account.slot, - signature: if let Some(txn_signature) = account_info.txn_signature { - Signature::try_from(txn_signature.as_slice()).expect("valid signature") - } else { - Signature::default() - }, - pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"), - executable: account_info.executable, - lamports: account_info.lamports, - owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), - rent_epoch: account_info.rent_epoch, - data: account_info.data, - program_received_time_us: chrono::Utc::now().timestamp_micros(), + slot: 0, + transaction_index: None, + block_hash: String::new(), + 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(), + }), + program_received_time_us: 0, } } } -impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { - fn from( - (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): ( - SubscribeUpdateBlockMeta, - Option, - ), - ) -> Self { - Self { - block_hash: blockhash.to_string(), - block_time, - slot, - program_received_time_us: chrono::Utc::now().timestamp_micros(), - } - } -} +// impl From for AccountPretty { +// fn from(account: SubscribeUpdateAccount) -> Self { +// let account_info = account.account.unwrap(); +// Self { +// slot: account.slot, +// signature: if let Some(txn_signature) = account_info.txn_signature { +// Signature::try_from(txn_signature.as_slice()).expect("valid signature") +// } else { +// Signature::default() +// }, +// pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"), +// executable: account_info.executable, +// lamports: account_info.lamports, +// owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), +// rent_epoch: account_info.rent_epoch, +// data: account_info.data, +// program_received_time_us: get_high_perf_clock(), +// } +// } +// } -impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { - fn from( - (SubscribeUpdateTransaction { transaction, slot }, block_time): ( - SubscribeUpdateTransaction, - Option, - ), - ) -> Self { - let tx = transaction.expect("should be defined"); - // 根据用户说明，交易索引在 transaction.index 中 - let transaction_index = tx.index; - Self { - slot, - transaction_index: Some(transaction_index), // 提取交易索引 - block_time, - block_hash: "".to_string(), - signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), - is_vote: tx.is_vote, - tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) - .expect("valid tx with meta"), - program_received_time_us: chrono::Utc::now().timestamp_micros(), - } - } -} +// impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { +// fn from( +// (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): ( +// SubscribeUpdateBlockMeta, +// Option, +// ), +// ) -> Self { +// Self { +// block_hash: blockhash, +// block_time, +// slot, +// program_received_time_us: get_high_perf_clock(), +// } +// } +// } + +// impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty { +// fn from( +// (SubscribeUpdateTransaction { transaction, slot }, block_time): ( +// SubscribeUpdateTransaction, +// Option, +// ), +// ) -> Self { +// let tx = transaction.expect("should be defined"); +// // 根据用户说明，交易索引在 transaction.index 中 +// let transaction_index = tx.index; +// Self { +// slot, +// transaction_index: Some(transaction_index), // 提取交易索引 +// block_time, +// block_hash: String::new(), +// signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), +// is_vote: tx.is_vote, +// tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) +// .expect("valid tx with meta"), +// program_received_time_us: get_high_perf_clock(), +// } +// } +// } diff --git a/src/streaming/shred/mod.rs b/src/streaming/shred/mod.rs index 1358fe9..cd7ce22 100644 --- a/src/streaming/shred/mod.rs +++ b/src/streaming/shred/mod.rs @@ -1,9 +1,11 @@ // ShredStream 相关模块 pub mod connection; +pub mod pool; pub mod types; // 重新导出主要类型 pub use connection::*; +pub use pool::*; pub use types::*; // 从公用模块重新导出 diff --git a/src/streaming/shred/pool.rs b/src/streaming/shred/pool.rs new file mode 100644 index 0000000..07e40df --- /dev/null +++ b/src/streaming/shred/pool.rs @@ -0,0 +1,156 @@ +use std::sync::{Arc, Mutex}; +use std::collections::VecDeque; +use std::ops::DerefMut; +use solana_sdk::transaction::VersionedTransaction; + +use super::TransactionWithSlot; + + +/// TransactionWithSlot 对象池 +pub struct TransactionWithSlotPool { + pool: Arc>>>, + max_size: usize, +} + +impl TransactionWithSlotPool { + pub fn new(initial_size: usize, max_size: usize) -> Self { + let mut pool = VecDeque::with_capacity(initial_size); + + // 预分配对象 + for _ in 0..initial_size { + pool.push_back(Box::new(TransactionWithSlot::default())); + } + + Self { pool: Arc::new(Mutex::new(pool)), max_size } + } + + pub fn acquire(&self) -> PooledTransactionWithSlot { + let mut pool = self.pool.lock().unwrap(); + let transaction = match pool.pop_front() { + Some(reused) => reused, + None => Box::new(TransactionWithSlot::default()), + }; + + PooledTransactionWithSlot { + transaction, + pool: Arc::clone(&self.pool), + max_size: self.max_size + } + } +} + +/// 带自动归还的 TransactionWithSlot +pub struct PooledTransactionWithSlot { + transaction: Box, + pool: Arc>>>, + max_size: usize, +} + +impl PooledTransactionWithSlot { + /// 从原始数据重置 + pub fn reset_from_data( + &mut self, + transaction: VersionedTransaction, + slot: u64, + program_received_time_us: i64 + ) { + self.transaction.transaction = transaction; + self.transaction.slot = slot; + self.transaction.program_received_time_us = program_received_time_us; + } + + /// 使用优化的工厂方法创建 TransactionWithSlot（移动数据而不是克隆） + pub fn into_transaction_with_slot(mut self) -> TransactionWithSlot { + // 移动数据而不是克隆，避免多余的内存分配 + std::mem::replace(self.deref_mut(), TransactionWithSlot::default()) + } +} + +impl Drop for PooledTransactionWithSlot { + fn drop(&mut self) { + let mut pool = self.pool.lock().unwrap(); + if pool.len() < self.max_size { + // 清理敏感数据 + self.transaction.slot = 0; + self.transaction.program_received_time_us = 0; + // 重置交易为默认值以清理敏感数据 + self.transaction.transaction = VersionedTransaction::default(); + pool.push_back(std::mem::take(&mut self.transaction)); + } + } +} + +impl std::ops::Deref for PooledTransactionWithSlot { + type Target = TransactionWithSlot; + + fn deref(&self) -> &Self::Target { + &self.transaction + } +} + +impl std::ops::DerefMut for PooledTransactionWithSlot { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.transaction + } +} + +/// Shred 对象池管理器 +pub struct ShredPoolManager { + transaction_pool: TransactionWithSlotPool, +} + +impl ShredPoolManager { + pub fn new() -> Self { + Self { + transaction_pool: TransactionWithSlotPool::new( + 5000, // 初始大小 - Shred 事件通常较多 + 15000, // 最大大小 + ), + } + } + + pub fn get_transaction_pool(&self) -> &TransactionWithSlotPool { + &self.transaction_pool + } + + /// 创建优化的 TransactionWithSlot + pub fn create_transaction_with_slot_optimized( + &self, + transaction: VersionedTransaction, + slot: u64, + program_received_time_us: i64, + ) -> TransactionWithSlot { + let mut pooled_tx = self.transaction_pool.acquire(); + pooled_tx.reset_from_data(transaction, slot, program_received_time_us); + pooled_tx.into_transaction_with_slot() + } +} + +impl Default for ShredPoolManager { + fn default() -> Self { + Self::new() + } +} + +// 全局 Shred 池管理器实例 +lazy_static::lazy_static! { + pub static ref GLOBAL_SHRED_POOL_MANAGER: ShredPoolManager = ShredPoolManager::new(); +} + +/// 便捷的全局工厂函数 +pub mod factory { + use super::*; + + /// 使用对象池创建 TransactionWithSlot（推荐用于高性能场景） + pub fn create_transaction_with_slot_pooled( + transaction: VersionedTransaction, + slot: u64, + program_received_time_us: i64, + ) -> TransactionWithSlot { + GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized( + transaction, + slot, + program_received_time_us + ) + } +} diff --git a/src/streaming/shred/types.rs b/src/streaming/shred/types.rs index d10094e..7e90d74 100644 --- a/src/streaming/shred/types.rs +++ b/src/streaming/shred/types.rs @@ -1,7 +1,7 @@ use solana_sdk::transaction::VersionedTransaction; /// 携带槽位信息的交易 -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub struct TransactionWithSlot { pub transaction: VersionedTransaction, pub slot: u64, diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 34101c0..7fe862a 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -8,7 +8,8 @@ use crate::protos::shredstream::SubscribeEntriesRequest; use crate::streaming::common::{EventProcessor, SubscriptionHandle}; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; -use crate::streaming::shred::TransactionWithSlot; +use crate::streaming::event_parser::core::traits::get_high_perf_clock; +use crate::streaming::shred::pool::factory; use log::error; use solana_entry::entry::Entry; @@ -57,10 +58,10 @@ impl ShredStreamGrpc { if let Ok(entries) = bincode::deserialize::>(&msg.entries) { for entry in entries { for transaction in entry.transactions { - let transaction_with_slot = TransactionWithSlot::new( + let transaction_with_slot = factory::create_transaction_with_slot_pooled( transaction.clone(), msg.slot, - chrono::Utc::now().timestamp_micros(), + get_high_perf_clock(), ); // 直接处理，背压控制在 EventProcessor 内部处理 if let Err(e) = event_processor_clone diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index f083b8b..bf30b31 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -5,8 +5,9 @@ use crate::streaming::common::{ use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::{Protocol, UnifiedEvent}; use crate::streaming::grpc::{ - AccountPretty, BlockMetaPretty, EventPretty, SubscriptionManager, TransactionPretty, + EventPretty, SubscriptionManager, }; +use crate::streaming::grpc::pool::factory; use anyhow::anyhow; use chrono::Local; use futures::channel::mpsc; @@ -224,7 +225,7 @@ impl YellowstoneGrpc { let created_at = msg.created_at; match msg.update_oneof { Some(UpdateOneof::Account(account)) => { - let account_pretty = AccountPretty::from(account); + let account_pretty = factory::create_account_pretty_pooled(account); log::debug!("Received account: {:?}", account_pretty); if let Err(e) = event_processor .process_grpc_event_transaction_with_metrics( @@ -237,8 +238,7 @@ impl YellowstoneGrpc { } } Some(UpdateOneof::BlockMeta(sut)) => { - let block_meta_pretty = - BlockMetaPretty::from((sut, created_at)); + let block_meta_pretty = factory::create_block_meta_pretty_pooled(sut, created_at); log::debug!("Received block meta: {:?}", block_meta_pretty); if let Err(e) = event_processor .process_grpc_event_transaction_with_metrics( @@ -251,8 +251,7 @@ impl YellowstoneGrpc { } } Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = - TransactionPretty::from((sut, created_at)); + let transaction_pretty = factory::create_transaction_pretty_pooled(sut, created_at); log::debug!( "Received transaction: {} at slot {}", transaction_pretty.signature, diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 030397d..9d0c710 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -1,9 +1,6 @@ use crate::{ common::AnyResult, - streaming::{ - grpc::{EventPretty, TransactionPretty}, - yellowstone_grpc::YellowstoneGrpc, - }, + streaming::{grpc::pool::factory, grpc::EventPretty, yellowstone_grpc::YellowstoneGrpc}, }; use futures::{SinkExt, StreamExt}; use log::error; @@ -62,13 +59,11 @@ impl YellowstoneGrpc { let created_at = msg.created_at; match msg.update_oneof { Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = TransactionPretty::from((sut, created_at)); + let transaction_pretty = + factory::create_transaction_pretty_pooled(sut, created_at); let event_pretty = EventPretty::Transaction(transaction_pretty); - if let Err(e) = Self::process_system_transaction( - event_pretty, - &*callback, - ) - .await + if let Err(e) = + Self::process_system_transaction(event_pretty, &*callback).await { error!("Error processing transaction: {e:?}"); } From 7601eb370cfbe1fda97a8fbbed061d6c477e6f83 Mon Sep 17 00:00:00 2001 From: ysq Date: Wed, 3 Sep 2025 15:29:57 +0800 Subject: [PATCH 2/4] 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), - })); } _ => {} } From 1060b04b129dab7226cc31dfa234269864abd2ce Mon Sep 17 00:00:00 2001 From: ysq Date: Wed, 3 Sep 2025 15:49:50 +0800 Subject: [PATCH 3/4] refactor: simplify event API method names for better ergonomics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rename verbose method names to shorter, clearer alternatives - program_received_time_us() → recv_us() - program_handle_time_consuming_us() → handle_us() - instruction_outer_index() → outer_index() - instruction_inner_index() → inner_index() - Update all implementations across event parsers and processors Improves developer experience while maintaining semantic clarity. --- Cargo.toml | 2 +- README.md | 4 +- README_CN.md | 4 +- src/main.rs | 2 +- src/streaming/common/event_processor.rs | 16 +-- src/streaming/event_parser/common/mod.rs | 20 ++-- src/streaming/event_parser/common/types.rs | 22 ++--- .../event_parser/core/account_event_parser.rs | 6 +- .../event_parser/core/common_event_parser.rs | 6 +- src/streaming/event_parser/core/macros.rs | 16 +-- src/streaming/event_parser/core/traits.rs | 98 +++++++++---------- .../protocols/block/block_meta_event.rs | 4 +- src/streaming/grpc/pool.rs | 6 +- src/streaming/grpc/types.rs | 18 ++-- src/streaming/shred/pool.rs | 14 +-- src/streaming/shred/types.rs | 6 +- 16 files changed, 122 insertions(+), 122 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index f53deae..7a7f89d 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.4.0" +version = "0.4.1" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index f45a700..d301572 100755 --- a/README.md +++ b/README.md @@ -46,14 +46,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.0" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.4.0" +solana-streamer-sdk = "0.4.1" ``` ## Configuration System diff --git a/README_CN.md b/README_CN.md index a88d3d7..9988650 100644 --- a/README_CN.md +++ b/README_CN.md @@ -45,14 +45,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.0" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.4.0" +solana-streamer-sdk = "0.4.1" ``` ## 配置系统 diff --git a/src/main.rs b/src/main.rs index d6ad5c3..5c3fd40 100755 --- a/src/main.rs +++ b/src/main.rs @@ -196,7 +196,7 @@ fn create_event_callback() -> impl Fn(Box) { match_event!(event, { // -------------------------- block meta ----------------------- BlockMetaEvent => |e: BlockMetaEvent| { - println!("BlockMetaEvent: {:?}", e.metadata.program_handle_time_consuming_us); + println!("BlockMetaEvent: {:?}", e.metadata.handle_us); }, // -------------------------- bonk ----------------------- BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index c25930f..dbe6517 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -94,7 +94,7 @@ impl EventProcessor { let metrics_manager = self.metrics_manager.clone(); Arc::new(move |event: Box| { - let processing_time_us = event.program_handle_time_consuming_us() as f64; + let processing_time_us = event.handle_us() as f64; callback(event); metrics_manager.update_metrics(MetricsEventType::Transaction, 1, processing_time_us); }) @@ -171,7 +171,7 @@ impl EventProcessor { self.event_type_filter.as_ref(), ); if let Some(event) = account_event { - let processing_time_us = event.program_handle_time_consuming_us() as f64; + let processing_time_us = event.handle_us() as f64; self.invoke_callback(event); self.update_metrics(MetricsEventType::Account, 1, processing_time_us); } @@ -181,7 +181,7 @@ impl EventProcessor { let slot = transaction_pretty.slot; let signature = transaction_pretty.signature; let block_time = transaction_pretty.block_time; - let program_received_time_us = transaction_pretty.program_received_time_us; + let recv_us = transaction_pretty.recv_us; let transaction_index = transaction_pretty.transaction_index; let grpc_tx = transaction_pretty.grpc_tx; @@ -193,7 +193,7 @@ impl EventProcessor { signature, Some(slot), block_time, - program_received_time_us, + recv_us, bot_wallet, transaction_index, adapter_callback, @@ -210,9 +210,9 @@ impl EventProcessor { block_meta_pretty.slot, block_meta_pretty.block_hash, block_time_ms, - block_meta_pretty.program_received_time_us, + block_meta_pretty.recv_us, ); - let processing_time_us = block_meta_event.program_handle_time_consuming_us() as f64; + let processing_time_us = block_meta_event.handle_us() as f64; self.invoke_callback(block_meta_event); self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us); } @@ -302,7 +302,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; + let recv_us = transaction_with_slot.recv_us; let parser = self.get_parser(); let adapter_callback = self.create_adapter_callback(); @@ -312,7 +312,7 @@ impl EventProcessor { signature, Some(slot), None, - program_received_time_us, + recv_us, bot_wallet, None, &[], diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index 6784e91..92424c6 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -20,16 +20,16 @@ macro_rules! impl_unified_event { self.metadata.slot } - fn program_received_time_us(&self) -> i64 { - self.metadata.program_received_time_us + fn recv_us(&self) -> i64 { + self.metadata.recv_us } - fn program_handle_time_consuming_us(&self) -> i64 { - self.metadata.program_handle_time_consuming_us + fn handle_us(&self) -> i64 { + self.metadata.handle_us } - fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64) { - self.metadata.program_handle_time_consuming_us = program_handle_time_consuming_us; + fn set_handle_us(&mut self, handle_us: i64) { + self.metadata.handle_us = handle_us; } fn as_any(&self) -> &dyn std::any::Any { @@ -60,12 +60,12 @@ macro_rules! impl_unified_event { self.metadata.swap_data.is_some() } - fn instruction_outer_index(&self) -> i64 { - self.metadata.instruction_outer_index + fn outer_index(&self) -> i64 { + self.metadata.outer_index } - fn instruction_inner_index(&self) -> Option { - self.metadata.instruction_inner_index + fn inner_index(&self) -> Option { + self.metadata.inner_index } fn transaction_index(&self) -> Option { self.metadata.transaction_index diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 877e940..888a088 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -293,14 +293,14 @@ pub struct EventMetadata { pub transaction_index: Option, // 新增：交易在slot中的索引 pub block_time: i64, pub block_time_ms: i64, - pub program_received_time_us: i64, - pub program_handle_time_consuming_us: i64, + pub recv_us: i64, + pub handle_us: i64, pub protocol: ProtocolType, pub event_type: EventType, pub program_id: Pubkey, pub swap_data: Option, - pub instruction_outer_index: i64, - pub instruction_inner_index: Option, + pub outer_index: i64, + pub inner_index: Option, } impl EventMetadata { @@ -313,9 +313,9 @@ impl EventMetadata { protocol: ProtocolType, event_type: EventType, program_id: Pubkey, - instruction_outer_index: i64, - instruction_inner_index: Option, - program_received_time_us: i64, + outer_index: i64, + inner_index: Option, + recv_us: i64, transaction_index: Option, ) -> Self { Self { @@ -323,14 +323,14 @@ impl EventMetadata { slot, block_time, block_time_ms, - program_received_time_us, - program_handle_time_consuming_us: 0, + recv_us, + handle_us: 0, protocol, event_type, program_id, swap_data: None, - instruction_outer_index, - instruction_inner_index, + outer_index, + inner_index, transaction_index, } } diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 001b263..a9926c5 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -184,13 +184,13 @@ impl AccountEventParser { protocol: config.protocol_type, event_type: config.event_type, program_id: config.program_id, - program_received_time_us: account.program_received_time_us, + recv_us: account.recv_us, ..Default::default() }, ); if let Some(mut event) = event { - event.set_program_handle_time_consuming_us(elapsed_micros_since( - account.program_received_time_us, + event.set_handle_us(elapsed_micros_since( + account.recv_us, )); return Some(event); } diff --git a/src/streaming/event_parser/core/common_event_parser.rs b/src/streaming/event_parser/core/common_event_parser.rs index 822d1e8..56f0e8f 100644 --- a/src/streaming/event_parser/core/common_event_parser.rs +++ b/src/streaming/event_parser/core/common_event_parser.rs @@ -8,12 +8,12 @@ impl CommonEventParser { slot: u64, block_hash: String, block_time_ms: i64, - program_received_time_us: i64, + recv_us: i64, ) -> Box { let mut block_meta_event = - BlockMetaEvent::new(slot, block_hash, block_time_ms, program_received_time_us); + BlockMetaEvent::new(slot, block_hash, block_time_ms, recv_us); block_meta_event - .set_program_handle_time_consuming_us(elapsed_micros_since(program_received_time_us)); + .set_handle_us(elapsed_micros_since(recv_us)); Box::new(block_meta_event) } } diff --git a/src/streaming/event_parser/core/macros.rs b/src/streaming/event_parser/core/macros.rs index 12dd7e2..0216bb7 100644 --- a/src/streaming/event_parser/core/macros.rs +++ b/src/streaming/event_parser/core/macros.rs @@ -30,7 +30,7 @@ macro_rules! impl_event_parser_delegate { signature: solana_sdk::signature::Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -41,7 +41,7 @@ macro_rules! impl_event_parser_delegate { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -55,13 +55,13 @@ macro_rules! impl_event_parser_delegate { signature: solana_sdk::signature::Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_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) + self.inner.parse_events_from_grpc_inner_instruction(inner_instruction, signature, slot, block_time, recv_us, outer_index, inner_index, transaction_index, config) } fn parse_events_from_instruction( @@ -71,7 +71,7 @@ macro_rules! impl_event_parser_delegate { signature: solana_sdk::signature::Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -89,7 +89,7 @@ macro_rules! impl_event_parser_delegate { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, bot_wallet, @@ -106,7 +106,7 @@ macro_rules! impl_event_parser_delegate { signature: solana_sdk::signature::Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -118,7 +118,7 @@ macro_rules! impl_event_parser_delegate { + 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) + self.inner.parse_events_from_grpc_instruction(instruction, accounts, signature, slot, block_time, recv_us, outer_index, inner_index, bot_wallet, transaction_index, inner_instructions, callback) } fn should_handle(&self, program_id: &solana_sdk::pubkey::Pubkey) -> bool { diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 8106a78..82ae0fc 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -236,13 +236,13 @@ pub trait UnifiedEvent: Debug + Send + Sync { fn slot(&self) -> u64; /// Get program received timestamp (milliseconds) - fn program_received_time_us(&self) -> i64; + fn recv_us(&self) -> i64; /// Processing time consumption (milliseconds) - fn program_handle_time_consuming_us(&self) -> i64; + fn handle_us(&self) -> i64; /// Set processing time consumption (milliseconds) - fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64); + fn set_handle_us(&mut self, handle_us: i64); /// Convert event to Any for downcasting fn as_any(&self) -> &dyn std::any::Any; @@ -265,8 +265,8 @@ pub trait UnifiedEvent: Debug + Send + Sync { fn swap_data_is_parsed(&self) -> bool; /// Get index - fn instruction_outer_index(&self) -> i64; - fn instruction_inner_index(&self) -> Option; + fn outer_index(&self) -> i64; + fn inner_index(&self) -> Option; /// Get transaction index in slot fn transaction_index(&self) -> Option; @@ -285,7 +285,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -300,7 +300,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -316,7 +316,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -335,7 +335,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -351,7 +351,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, accounts: &[Pubkey], inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions], bot_wallet: Option, @@ -383,7 +383,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, index as i64, None, bot_wallet, @@ -407,7 +407,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, accounts: &[Pubkey], inner_instructions: &[InnerInstructions], bot_wallet: Option, @@ -440,7 +440,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, index as i64, None, bot_wallet, @@ -462,7 +462,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, bot_wallet: Option, transaction_index: Option, inner_instructions: &[InnerInstructions], @@ -477,7 +477,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, bot_wallet, transaction_index, inner_instructions, @@ -493,7 +493,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, bot_wallet: Option, transaction_index: Option, inner_instructions: &[InnerInstructions], @@ -505,7 +505,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, &accounts, inner_instructions, bot_wallet, @@ -522,7 +522,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, bot_wallet: Option, transaction_index: Option, callback: Arc) + Send + Sync>, @@ -537,7 +537,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, bot_wallet, transaction_index, adapter_callback, @@ -551,7 +551,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, bot_wallet: Option, transaction_index: Option, callback: Arc Fn(&'a Box) + Send + Sync>, @@ -600,7 +600,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, &accounts_arc, &inner_instructions_arc, bot_wallet, @@ -625,7 +625,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, inner_instruction.index as i64, Some(index as i64), bot_wallet, @@ -732,7 +732,7 @@ pub trait EventParser: Send + Sync { let slot = transaction.slot; let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 }); - let program_received_time_us = get_high_perf_clock(); + let recv_us = get_high_perf_clock(); let bot_wallet = None; let transaction_index = None; // 解析指令事件 @@ -741,7 +741,7 @@ pub trait EventParser: Send + Sync { signature, Some(slot), block_time, - program_received_time_us, + recv_us, &accounts_arc, &inner_instructions_arc, bot_wallet, @@ -759,7 +759,7 @@ pub trait EventParser: Send + Sync { signature, Some(slot), block_time, - program_received_time_us, + recv_us, inner_instruction.index as i64, Some(index as i64), bot_wallet, @@ -780,7 +780,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -792,7 +792,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -809,7 +809,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -824,7 +824,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, bot_wallet, @@ -842,7 +842,7 @@ pub trait EventParser: Send + Sync { signature: Signature, slot: Option, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -857,7 +857,7 @@ pub trait EventParser: Send + Sync { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, bot_wallet, @@ -938,7 +938,7 @@ impl GenericEventParser { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -956,7 +956,7 @@ impl GenericEventParser { config.program_id, outer_index, inner_index, - program_received_time_us, + recv_us, transaction_index, ); parser(data, metadata) @@ -975,7 +975,7 @@ impl GenericEventParser { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -993,7 +993,7 @@ impl GenericEventParser { config.program_id, outer_index, inner_index, - program_received_time_us, + recv_us, transaction_index, ); parser(data, account_pubkeys, metadata) @@ -1016,7 +1016,7 @@ impl EventParser for GenericEventParser { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -1034,7 +1034,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -1052,7 +1052,7 @@ impl EventParser for GenericEventParser { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, transaction_index: Option, @@ -1070,7 +1070,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -1089,7 +1089,7 @@ impl EventParser for GenericEventParser { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -1141,7 +1141,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -1165,7 +1165,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -1205,8 +1205,8 @@ impl EventParser for GenericEventParser { event.merge(&*inner_instruction_event); } // 设置处理时间（使用高性能时钟） - event.set_program_handle_time_consuming_us(elapsed_micros_since( - program_received_time_us, + event.set_handle_us(elapsed_micros_since( + recv_us, )); event = process_event(event, bot_wallet); callback(&event); @@ -1224,7 +1224,7 @@ impl EventParser for GenericEventParser { signature: Signature, slot: u64, block_time: Option, - program_received_time_us: i64, + recv_us: i64, outer_index: i64, inner_index: Option, bot_wallet: Option, @@ -1276,7 +1276,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -1300,7 +1300,7 @@ impl EventParser for GenericEventParser { signature, slot, block_time, - program_received_time_us, + recv_us, outer_index, inner_index, transaction_index, @@ -1340,8 +1340,8 @@ impl EventParser for GenericEventParser { event.merge(&*inner_instruction_event); } // 设置处理时间（使用高性能时钟） - event.set_program_handle_time_consuming_us(elapsed_micros_since( - program_received_time_us, + event.set_handle_us(elapsed_micros_since( + recv_us, )); event = process_event(event, bot_wallet); callback(&event); diff --git a/src/streaming/event_parser/protocols/block/block_meta_event.rs b/src/streaming/event_parser/protocols/block/block_meta_event.rs index 392e24b..59390ce 100644 --- a/src/streaming/event_parser/protocols/block/block_meta_event.rs +++ b/src/streaming/event_parser/protocols/block/block_meta_event.rs @@ -18,7 +18,7 @@ impl BlockMetaEvent { slot: u64, block_hash: String, block_time_ms: i64, - program_received_time_us: i64, + recv_us: i64, ) -> Self { let metadata = EventMetadata::new( Signature::default(), @@ -30,7 +30,7 @@ impl BlockMetaEvent { solana_sdk::pubkey::Pubkey::default(), 0, None, - program_received_time_us, + recv_us, None, ); Self { metadata, slot, block_hash } diff --git a/src/streaming/grpc/pool.rs b/src/streaming/grpc/pool.rs index afc2082..56627c0 100644 --- a/src/streaming/grpc/pool.rs +++ b/src/streaming/grpc/pool.rs @@ -118,7 +118,7 @@ impl PooledAccountPretty { self.account.data = new_data; } - self.account.program_received_time_us = get_high_perf_clock(); + self.account.recv_us = get_high_perf_clock(); } } @@ -196,7 +196,7 @@ impl PooledBlockMetaPretty { self.block_meta.slot = block_update.slot; self.block_meta.block_hash = block_update.blockhash; self.block_meta.block_time = block_time; - self.block_meta.program_received_time_us = get_high_perf_clock(); + self.block_meta.recv_us = get_high_perf_clock(); } } @@ -282,7 +282,7 @@ impl PooledTransactionPretty { self.transaction.signature = Signature::try_from(tx.signature.as_slice()).expect("valid signature"); self.transaction.is_vote = tx.is_vote; - self.transaction.program_received_time_us = get_high_perf_clock(); + self.transaction.recv_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 7f2e771..db7fe1c 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -29,7 +29,7 @@ pub struct AccountPretty { pub owner: Pubkey, pub rent_epoch: u64, pub data: Vec, - pub program_received_time_us: i64, + pub recv_us: i64, } impl fmt::Debug for AccountPretty { @@ -52,7 +52,7 @@ pub struct BlockMetaPretty { pub slot: u64, pub block_hash: String, pub block_time: Option, - pub program_received_time_us: i64, + pub recv_us: i64, } impl fmt::Debug for BlockMetaPretty { @@ -61,7 +61,7 @@ impl fmt::Debug for BlockMetaPretty { .field("slot", &self.slot) .field("block_hash", &self.block_hash) .field("block_time", &self.block_time) - .field("program_received_time_us", &self.program_received_time_us) + .field("recv_us", &self.recv_us) .finish() } } @@ -74,7 +74,7 @@ pub struct TransactionPretty { pub block_time: Option, pub signature: Signature, pub is_vote: bool, - pub program_received_time_us: i64, + pub recv_us: i64, pub grpc_tx: SubscribeUpdateTransactionInfo, } @@ -85,7 +85,7 @@ impl fmt::Debug for TransactionPretty { .field("transaction_index", &self.transaction_index) .field("signature", &self.signature) .field("is_vote", &self.is_vote) - .field("program_received_time_us", &self.program_received_time_us) + .field("recv_us", &self.recv_us) .finish() } } @@ -100,7 +100,7 @@ impl Default for TransactionPretty { signature: Signature::default(), is_vote: false, grpc_tx: SubscribeUpdateTransactionInfo::default(), - program_received_time_us: 0, + recv_us: 0, } } } @@ -121,7 +121,7 @@ impl Default for TransactionPretty { // owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"), // rent_epoch: account_info.rent_epoch, // data: account_info.data, -// program_received_time_us: get_high_perf_clock(), +// recv_us: get_high_perf_clock(), // } // } // } @@ -137,7 +137,7 @@ impl Default for TransactionPretty { // block_hash: blockhash, // block_time, // slot, -// program_received_time_us: get_high_perf_clock(), +// recv_us: get_high_perf_clock(), // } // } // } @@ -161,7 +161,7 @@ impl Default for TransactionPretty { // is_vote: tx.is_vote, // tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) // .expect("valid tx with meta"), -// program_received_time_us: get_high_perf_clock(), +// recv_us: get_high_perf_clock(), // } // } // } diff --git a/src/streaming/shred/pool.rs b/src/streaming/shred/pool.rs index 07e40df..5dea151 100644 --- a/src/streaming/shred/pool.rs +++ b/src/streaming/shred/pool.rs @@ -52,11 +52,11 @@ impl PooledTransactionWithSlot { &mut self, transaction: VersionedTransaction, slot: u64, - program_received_time_us: i64 + recv_us: i64 ) { self.transaction.transaction = transaction; self.transaction.slot = slot; - self.transaction.program_received_time_us = program_received_time_us; + self.transaction.recv_us = recv_us; } /// 使用优化的工厂方法创建 TransactionWithSlot（移动数据而不是克隆） @@ -72,7 +72,7 @@ impl Drop for PooledTransactionWithSlot { if pool.len() < self.max_size { // 清理敏感数据 self.transaction.slot = 0; - self.transaction.program_received_time_us = 0; + self.transaction.recv_us = 0; // 重置交易为默认值以清理敏感数据 self.transaction.transaction = VersionedTransaction::default(); pool.push_back(std::mem::take(&mut self.transaction)); @@ -118,10 +118,10 @@ impl ShredPoolManager { &self, transaction: VersionedTransaction, slot: u64, - program_received_time_us: i64, + recv_us: i64, ) -> TransactionWithSlot { let mut pooled_tx = self.transaction_pool.acquire(); - pooled_tx.reset_from_data(transaction, slot, program_received_time_us); + pooled_tx.reset_from_data(transaction, slot, recv_us); pooled_tx.into_transaction_with_slot() } } @@ -145,12 +145,12 @@ pub mod factory { pub fn create_transaction_with_slot_pooled( transaction: VersionedTransaction, slot: u64, - program_received_time_us: i64, + recv_us: i64, ) -> TransactionWithSlot { GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized( transaction, slot, - program_received_time_us + recv_us ) } } diff --git a/src/streaming/shred/types.rs b/src/streaming/shred/types.rs index 7e90d74..ca87cc4 100644 --- a/src/streaming/shred/types.rs +++ b/src/streaming/shred/types.rs @@ -5,7 +5,7 @@ use solana_sdk::transaction::VersionedTransaction; pub struct TransactionWithSlot { pub transaction: VersionedTransaction, pub slot: u64, - pub program_received_time_us: i64, + pub recv_us: i64, } impl TransactionWithSlot { @@ -13,8 +13,8 @@ impl TransactionWithSlot { pub fn new( transaction: VersionedTransaction, slot: u64, - program_received_time_us: i64, + recv_us: i64, ) -> Self { - Self { transaction, slot, program_received_time_us } + Self { transaction, slot, recv_us } } } From 4902c34bfabf081c614065f788b9bd4e649731b6 Mon Sep 17 00:00:00 2001 From: ysq Date: Wed, 3 Sep 2025 17:19:35 +0800 Subject: [PATCH 4/4] feat: add common account event parser with SPL token support --- Cargo.toml | 1 + src/main.rs | 4 + src/streaming/event_parser/common/types.rs | 3 + .../event_parser/core/account_event_parser.rs | 92 +++++++++++++++---- 4 files changed, 83 insertions(+), 17 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 7a7f89d..0e61ef4 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,6 +68,7 @@ crossbeam = "0.8.4" crossbeam-queue = "0.3.12" parking_lot = "0.12.1" wide = "0.7" +spl-token = "8.0.0" [dev-dependencies] criterion = { version = "0.5", features = ["html_reports"] } \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 5c3fd40..c28031d 100755 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,7 @@ use solana_streamer_sdk::{ streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventType}, + core::account_event_parser::CommonAccountEvent, protocols::{ bonk::{ parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, @@ -333,6 +334,9 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| { println!("RaydiumCpmmPoolStateAccountEvent: {e:?}"); }, + CommonAccountEvent => |e: CommonAccountEvent| { + println!("CommonAccountEvent: {e:?}"); + }, }); } } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 888a088..b935557 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -141,6 +141,8 @@ pub enum EventType { AccountRaydiumCpmmAmmConfig, AccountRaydiumCpmmPoolState, + AccountCommon, + // Common events BlockMeta, Unknown, @@ -225,6 +227,7 @@ impl fmt::Display for EventType { } EventType::AccountRaydiumCpmmAmmConfig => write!(f, "AccountRaydiumCpmmAmmConfig"), EventType::AccountRaydiumCpmmPoolState => write!(f, "AccountRaydiumCpmmPoolState"), + EventType::AccountCommon => write!(f, "AccountCommon"), EventType::BlockMeta => write!(f, "BlockMeta"), EventType::Unknown => write!(f, "Unknown"), } diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index a9926c5..4ac5c9a 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -1,8 +1,12 @@ use std::collections::HashMap; use std::sync::OnceLock; +use serde::{Deserialize, Serialize}; +use solana_sdk::program_pack::Pack; use solana_sdk::pubkey::Pubkey; +use spl_token::state::Account; +use crate::impl_unified_event; use crate::streaming::common::SimdUtils; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; @@ -26,6 +30,19 @@ pub struct AccountEventParseConfig { pub account_parser: AccountEventParserFn, } +/// 通用账户事件 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct CommonAccountEvent { + pub metadata: EventMetadata, + pub pubkey: Pubkey, + pub executable: bool, + pub lamports: u64, + pub owner: Pubkey, + pub rent_epoch: u64, + pub amount: Option, +} +impl_unified_event!(CommonAccountEvent,); + /// 账户事件解析器 pub type AccountEventParserFn = fn(account: &AccountPretty, metadata: EventMetadata) -> Option>; @@ -33,6 +50,9 @@ pub type AccountEventParserFn = static PROTOCOL_CONFIGS_CACHE: OnceLock>> = OnceLock::new(); +// 通用账户解析配置的静态缓存 +static COMMON_CONFIG: OnceLock = OnceLock::new(); + pub struct AccountEventParser {} impl AccountEventParser { @@ -148,21 +168,39 @@ impl AccountEventParser { map }); - let mut configs = vec![]; - let empty_vec = vec![]; + let mut configs = Vec::new(); + let empty_vec = Vec::new(); + + // 预估容量以减少重新分配 + let estimated_capacity = protocols.len() * 3; // 大多数协议有2-3个配置 + configs.reserve(estimated_capacity); + for protocol in protocols { let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec); - let filtered_configs: Vec = protocol_configs - .iter() - .filter(|config| { - event_type_filter - .map(|filter| filter.include.contains(&config.event_type)) - .unwrap_or(true) - }) - .cloned() - .collect(); - configs.extend(filtered_configs); + // 如果没有过滤器，直接扩展所有配置 + if event_type_filter.is_none() { + configs.extend(protocol_configs.iter().cloned()); + } else { + // 有过滤器时才进行过滤 + let filter = event_type_filter.unwrap(); + configs.extend( + protocol_configs + .iter() + .filter(|config| filter.include.contains(&config.event_type)) + .cloned(), + ); + } } + + let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig { + program_id: Pubkey::default(), + protocol_type: ProtocolType::Common, + event_type: EventType::AccountCommon, + account_discriminator: &[], + account_parser: Self::parse_token_account_event, + }); + configs.push(common_config.clone()); + configs } @@ -173,8 +211,12 @@ impl AccountEventParser { ) -> Option> { let configs = Self::configs(protocols, event_type_filter); for config in configs { - if account.owner == config.program_id - && SimdUtils::fast_discriminator_match(&account.data, config.account_discriminator) + if config.program_id == Pubkey::default() + || (account.owner == config.program_id + && SimdUtils::fast_discriminator_match( + &account.data, + config.account_discriminator, + )) { let event = (config.account_parser)( &account, @@ -189,13 +231,29 @@ impl AccountEventParser { }, ); if let Some(mut event) = event { - event.set_handle_us(elapsed_micros_since( - account.recv_us, - )); + event.set_handle_us(elapsed_micros_since(account.recv_us)); return Some(event); } } } None } + + pub fn parse_token_account_event( + account: &AccountPretty, + metadata: EventMetadata, + ) -> Option> { + let info = Account::unpack(&account.data); + let mut event = CommonAccountEvent { + metadata, + pubkey: account.pubkey, + executable: account.executable, + lamports: account.lamports, + owner: account.owner, + rent_epoch: account.rent_epoch, + amount: if let Ok(info) = info { Some(info.amount) } else { None }, + }; + event.set_handle_us(elapsed_micros_since(account.recv_us)); + return Some(Box::new(event)); + } }