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

perf: add object pooling and enhanced metrics for streaming optimization

Browse Source 
- 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
This commit is contained in:
ysq
2025-09-02 17:27:27 +08:00
parent 07bb110dc0
commit b71a83bf66
30 changed files with 920 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/streaming/common/config.rs
+1 -1
View File
@@ -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() }
}
}
src/streaming/common/event_processor.rs
+3 -3
View File
@@ -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,
);
src/streaming/common/metrics.rs
+42 -86
View File
@@ -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!();
}
src/streaming/event_parser/common/mod.rs
+1 -1
View File
@@ -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
}
src/streaming/event_parser/common/types.rs
+9 -11
View File
@@ -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<String>,
pub description: Option<Cow<'static, str>>,
}
/// 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<u64>, // 新增：交易在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 => {
src/streaming/event_parser/core/account_event_parser.rs
+5 -7
View File
@@ -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);
}
}
src/streaming/event_parser/core/common_event_parser.rs
+6 -11
View File
@@ -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<dyn UnifiedEvent> {
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)
}
}
src/streaming/event_parser/core/traits.rs
+95 -17
View File
@@ -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<HighPerformanceClock> =
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<u64>,
) -> Option<Box<dyn UnifiedEvent>> {
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<u64>,
) -> Option<Box<dyn UnifiedEvent>> {
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);
}
src/streaming/event_parser/protocols/block/block_meta_event.rs
+2 -3
View File
@@ -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,
src/streaming/event_parser/protocols/bonk/events.rs
+7 -6
View File
@@ -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,
}
src/streaming/event_parser/protocols/bonk/types.rs
+6 -6
View File
@@ -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,
}))
src/streaming/event_parser/protocols/pumpfun/events.rs
+4 -4
View File
@@ -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,
}
src/streaming/event_parser/protocols/pumpfun/types.rs
+4 -4
View File
@@ -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,
}))
src/streaming/event_parser/protocols/pumpswap/events.rs
+6 -4
View File
@@ -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,
}
src/streaming/event_parser/protocols/pumpswap/types.rs
+4 -4
View File
@@ -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,
}))
src/streaming/event_parser/protocols/raydium_amm_v4/events.rs
+8 -2
View File
@@ -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,
}
src/streaming/event_parser/protocols/raydium_amm_v4/types.rs
+2 -2
View File
@@ -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,
}))
src/streaming/event_parser/protocols/raydium_clmm/events.rs
+6 -6
View File
@@ -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,
}
src/streaming/event_parser/protocols/raydium_clmm/types.rs
+6 -6
View File
@@ -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,
}))
src/streaming/event_parser/protocols/raydium_cpmm/events.rs
+10 -4
View File
@@ -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,
}
src/streaming/event_parser/protocols/raydium_cpmm/types.rs
+4 -4
View File
@@ -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,
}))
src/streaming/grpc/mod.rs
+2
View File
@@ -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::*;
src/streaming/grpc/pool.rs
+438
View File
@@ -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<T> {
fn acquire(&self) -> PooledObject<T>;
fn return_object(&self, obj: Box<T>);
}
/// 带自动归还的智能指针
pub struct PooledObject<T> {
object: Option<Box<T>>,
pool: Arc<Mutex<VecDeque<Box<T>>>>,
max_size: usize,
}
impl<T> PooledObject<T> {
fn new(object: Box<T>, pool: Arc<Mutex<VecDeque<Box<T>>>>, max_size: usize) -> Self {
Self { object: Some(object), pool, max_size }
}
}
impl<T> Drop for PooledObject<T> {
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<T> std::ops::Deref for PooledObject<T> {
type Target = T;
fn deref(&self) -> &Self::Target {
self.object.as_ref().unwrap()
}
}
impl<T> std::ops::DerefMut for PooledObject<T> {
fn deref_mut(&mut self) -> &mut Self::Target {
self.object.as_mut().unwrap()
}
}
/// AccountPretty 对象池
pub struct AccountPrettyPool {
pool: Arc<Mutex<VecDeque<Box<AccountPretty>>>>,
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<AccountPretty>,
pool: Arc<Mutex<VecDeque<Box<AccountPretty>>>>,
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<Mutex<VecDeque<Box<BlockMetaPretty>>>>,
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<BlockMetaPretty>,
pool: Arc<Mutex<VecDeque<Box<BlockMetaPretty>>>>,
max_size: usize,
}
impl PooledBlockMetaPretty {
/// 从 gRPC 更新重置数据
pub fn reset_from_update(
&mut self,
block_update: SubscribeUpdateBlockMeta,
block_time: Option<Timestamp>,
) {
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<Mutex<VecDeque<Box<TransactionPretty>>>>,
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<TransactionPretty>,
pool: Arc<Mutex<VecDeque<Box<TransactionPretty>>>>,
max_size: usize,
}
impl PooledTransactionPretty {
/// 从 gRPC 更新重置数据
pub fn reset_from_update(
&mut self,
tx_update: SubscribeUpdateTransaction,
block_time: Option<Timestamp>,
) {
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<Timestamp>,
) -> 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<Timestamp>,
) -> 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<Timestamp>,
) -> 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<Timestamp>,
) -> TransactionPretty {
GLOBAL_POOL_MANAGER.get_event_pool().create_transaction_event_optimized(update, block_time)
}
}
src/streaming/grpc/types.rs
+76 -61
View File
@@ -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<SubscribeUpdateAccount> 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<Timestamp>)> for BlockMetaPretty {
fn from(
(SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): (
SubscribeUpdateBlockMeta,
Option<Timestamp>,
),
) -> Self {
Self {
block_hash: blockhash.to_string(),
block_time,
slot,
program_received_time_us: chrono::Utc::now().timestamp_micros(),
}
}
}
// impl From<SubscribeUpdateAccount> 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<Timestamp>)> for TransactionPretty {
fn from(
(SubscribeUpdateTransaction { transaction, slot }, block_time): (
SubscribeUpdateTransaction,
Option<Timestamp>,
),
) -> 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<Timestamp>)> for BlockMetaPretty {
// fn from(
// (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): (
// SubscribeUpdateBlockMeta,
// Option<Timestamp>,
// ),
// ) -> Self {
// Self {
// block_hash: blockhash,
// block_time,
// slot,
// program_received_time_us: get_high_perf_clock(),
// }
// }
// }
// impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty {
// fn from(
// (SubscribeUpdateTransaction { transaction, slot }, block_time): (
// SubscribeUpdateTransaction,
// Option<Timestamp>,
// ),
// ) -> 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(),
// }
// }
// }
src/streaming/shred/mod.rs
+2
View File
@@ -1,9 +1,11 @@
// ShredStream 相关模块
pub mod connection;
pub mod pool;
pub mod types;
// 重新导出主要类型
pub use connection::*;
pub use pool::*;
pub use types::*;
// 从公用模块重新导出
src/streaming/shred/pool.rs
+156
View File
@@ -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<Mutex<VecDeque<Box<TransactionWithSlot>>>>,
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<TransactionWithSlot>,
pool: Arc<Mutex<VecDeque<Box<TransactionWithSlot>>>>,
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
)
}
}
src/streaming/shred/types.rs
+1 -1
View File
@@ -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,
src/streaming/shred_stream.rs
+4 -3
View File
@@ -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::<Vec<Entry>>(&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
src/streaming/yellowstone_grpc.rs
+5 -6
View File
@@ -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,
src/streaming/yellowstone_sub_system.rs
+5 -10
View File
@@ -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:?}");
}