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

perf: Major event processing system refactor for improved performance

Browse Source
This commit is contained in:
ysq
2025-08-27 21:31:31 +08:00
parent e673b0aab8
commit 9ea4dab4df
22 changed files with 829 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.rs
+12 -14
View File
@@ -86,13 +86,17 @@ async fn test_grpc() -> Result<(), Box<dyn std::error::Error>> {
println!("Protocols to monitor: {:?}", protocols);
const SYSTEM_PROGRAM_ID: solana_sdk::pubkey::Pubkey =
solana_sdk::pubkey!("11111111111111111111111111111111");
// Filter accounts
let account_include = vec![
PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID
PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID
BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID
RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID
RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID
SYSTEM_PROGRAM_ID.to_string(),
PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID
PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID
BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID
RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID
RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID
RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // Listen to raydium_amm_v4 program ID
];
let account_exclude = vec![];
@@ -151,7 +155,7 @@ async fn test_shreds() -> Result<(), Box<dyn std::error::Error>> {
// Enable performance monitoring, has performance overhead, disabled by default
config.enable_metrics = true;
let shred_stream =
ShredStreamGrpc::new_with_config("http://127.0.0.1:10800".to_string(), config).await?;
ShredStreamGrpc::new_with_config("http://64.130.37.195:10800".to_string(), config).await?;
let callback = create_event_callback();
let protocols = vec![
@@ -188,17 +192,11 @@ async fn test_shreds() -> Result<(), Box<dyn std::error::Error>> {
fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
|event: Box<dyn UnifiedEvent>| {
println!(
"🎉 Event received! Type: {:?}, ID: {}, Slot: {}, Transaction Index: {:?}",
event.event_type(),
event.id(),
event.slot(),
event.transaction_index(),
);
println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id());
match_event!(event, {
// -------------------------- block meta -----------------------
BlockMetaEvent => |e: BlockMetaEvent| {
println!("BlockMetaEvent: {e:?}");
println!("BlockMetaEvent: {:?}", e.metadata.program_handle_time_consuming_us);
},
// -------------------------- bonk -----------------------
BonkPoolCreateEvent => |e: BonkPoolCreateEvent| {
src/streaming/common/constants.rs
+1 -1
View File
@@ -11,4 +11,4 @@ pub const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5;
// 性能监控相关常量
pub const DEFAULT_METRICS_WINDOW_SECONDS: u64 = 5;
pub const DEFAULT_METRICS_PRINT_INTERVAL_SECONDS: u64 = 10;
pub const SLOW_PROCESSING_THRESHOLD_US: f64 = 500.0;
pub const SLOW_PROCESSING_THRESHOLD_US: f64 = 3000.0;
src/streaming/common/event_processor.rs
+71 -73
View File
@@ -1,7 +1,7 @@
use std::sync::Arc;
use tokio::task::JoinHandle;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::Signature;
use crate::common::AnyResult;
use crate::streaming::common::{
@@ -14,7 +14,7 @@ use crate::streaming::event_parser::EventParser;
use crate::streaming::event_parser::{
core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol,
};
use crate::streaming::grpc::{BackpressureStrategy, BatchConfig, EventPretty};
use crate::streaming::grpc::{BackpressureConfig, BatchConfig, EventPretty};
use crate::streaming::shred::TransactionWithSlot;
use once_cell::sync::OnceCell;
@@ -25,21 +25,24 @@ pub struct EventProcessor {
pub(crate) parser_cache: OnceCell<Arc<dyn EventParser>>,
pub(crate) protocols: Vec<Protocol>,
pub(crate) event_type_filter: Option<EventTypeFilter>,
pub(crate) backpressure_strategy: BackpressureStrategy,
pub(crate) callback: Option<Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>>,
pub(crate) backpressure_config: BackpressureConfig,
pub(crate) batch_config: BatchConfig,
}
impl EventProcessor {
/// 创建新的事件处理器
pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self {
let backpressure_config = config.backpressure.clone();
Self {
metrics_manager,
config,
parser_cache: OnceCell::new(),
protocols: vec![],
event_type_filter: None,
backpressure_strategy: BackpressureStrategy::Block,
backpressure_config,
batch_config: BatchConfig::default(),
callback: None,
}
}
@@ -47,13 +50,15 @@ impl EventProcessor {
&mut self,
protocols: Vec<Protocol>,
event_type_filter: Option<EventTypeFilter>,
backpressure_strategy: BackpressureStrategy,
backpressure_config: BackpressureConfig,
batch_config: BatchConfig,
callback: Option<Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>>,
) {
self.protocols = protocols.clone();
self.event_type_filter = event_type_filter.clone();
self.backpressure_strategy = backpressure_strategy;
self.backpressure_config = backpressure_config;
self.batch_config = batch_config;
self.callback = callback;
self.parser_cache
.get_or_init(|| Arc::new(MutilEventParser::new(protocols, event_type_filter)));
}
@@ -62,35 +67,24 @@ impl EventProcessor {
self.parser_cache.get().unwrap().clone()
}
pub fn get_event_handle(&self) -> Option<JoinHandle<()>> {
return None;
}
pub async fn process_grpc_event_transaction_with_metrics<F>(
pub async fn process_grpc_event_transaction_with_metrics(
&self,
event_pretty: EventPretty,
callback: &F,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()>
where
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync,
{
self.process_grpc_event_transaction(event_pretty, callback, bot_wallet).await?;
) -> AnyResult<()> {
self.process_grpc_event_transaction(event_pretty, bot_wallet).await?;
Ok(())
}
async fn process_grpc_event_transaction<F>(
async fn process_grpc_event_transaction(
&self,
event_pretty: EventPretty,
callback: &F,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()>
where
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync,
{
) -> AnyResult<()> {
match event_pretty {
EventPretty::Account(account_pretty) => {
self.metrics_manager.add_account_process_count();
let signature = account_pretty.signature;
let account_event = AccountEventParser::parse_account_event(
self.protocols.clone(),
account_pretty,
@@ -98,12 +92,12 @@ impl EventProcessor {
);
if let Some(event) = account_event {
let processing_time_us = event.program_handle_time_consuming_us() as f64;
callback(event);
// 更新性能指标（如果启用）
self.metrics_manager.update_metrics(
self.invoke_callback(event);
self.update_metrics(
MetricsEventType::Account,
1,
processing_time_us,
Some(signature),
);
}
}
@@ -113,7 +107,7 @@ impl EventProcessor {
let signature = transaction_pretty.signature;
// 使用缓存获取解析器
let parser = self.get_parser();
let mut all_events = parser
let all_events = parser
.parse_transaction(
transaction_pretty.tx.clone(),
signature,
@@ -125,37 +119,26 @@ impl EventProcessor {
.await
.unwrap_or_else(|_e| vec![]);
// 为所有事件设置交易索引
for event in &mut all_events {
event.set_transaction_index(transaction_pretty.transaction_index);
}
let max_time_consuming_us = all_events
.iter()
.map(|event| event.program_handle_time_consuming_us())
.max()
.unwrap_or(0);
// 保存事件数量用于日志记录
let mut max_time_consuming_us = 0;
let event_count = all_events.len();
// 批量处理事件
if !all_events.is_empty() {
for mut event in all_events {
event.set_program_handle_time_consuming_us(
chrono::Utc::now().timestamp_micros()
- event.program_received_time_us(),
);
callback(event);
}
// 为所有事件设置交易索引
for mut event in all_events {
event.set_transaction_index(transaction_pretty.transaction_index);
event.set_program_handle_time_consuming_us(
chrono::Utc::now().timestamp_micros() - event.program_received_time_us(),
);
max_time_consuming_us =
max_time_consuming_us.max(event.program_handle_time_consuming_us());
self.invoke_callback(event);
}
// 更新性能指标
// 更新性能指标（如果启用）
self.metrics_manager.update_metrics(
MetricsEventType::Tx,
self.update_metrics(
MetricsEventType::Transaction,
event_count as u64,
max_time_consuming_us as f64,
Some(signature),
);
}
EventPretty::BlockMeta(block_meta_pretty) => {
@@ -171,29 +154,34 @@ impl EventProcessor {
block_meta_pretty.program_received_time_us,
);
let processing_time_us = block_meta_event.program_handle_time_consuming_us() as f64;
callback(block_meta_event);
// 更新性能指标（如果启用）
self.metrics_manager.update_metrics(
MetricsEventType::BlockMeta,
1,
processing_time_us,
);
self.invoke_callback(block_meta_event);
self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us, None);
}
}
Ok(())
}
pub fn invoke_callback(&self, event: Box<dyn UnifiedEvent>) {
if let Some(callback) = self.callback.as_ref() {
callback(event);
}
}
/// 即时处理单个交易
pub async fn process_shred_transaction_immediate<F>(
pub async fn process_shred_transaction_immediate(
&self,
transaction_with_slot: TransactionWithSlot,
bot_wallet: Option<Pubkey>,
callback: &F,
) -> AnyResult<()>
where
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync,
{
) -> AnyResult<()> {
self.process_shred_transaction(transaction_with_slot, bot_wallet).await
}
pub async fn process_shred_transaction(
&self,
transaction_with_slot: TransactionWithSlot,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()> {
self.metrics_manager.add_tx_process_count();
let program_received_time_us = chrono::Utc::now().timestamp_micros();
let slot = transaction_with_slot.slot;
@@ -215,11 +203,7 @@ impl EventProcessor {
.await
.unwrap_or_else(|_e| vec![]);
let max_time_consuming_us = all_events
.iter()
.map(|event| event.program_handle_time_consuming_us())
.max()
.unwrap_or(0);
let mut max_time_consuming_us = 0;
// 保存事件数量用于日志记录
let event_count = all_events.len();
@@ -229,18 +213,31 @@ impl EventProcessor {
event.set_program_handle_time_consuming_us(
chrono::Utc::now().timestamp_micros() - event.program_received_time_us(),
);
callback(event);
max_time_consuming_us =
max_time_consuming_us.max(event.program_handle_time_consuming_us());
self.invoke_callback(event);
}
// 实际调用性能指标更新
self.metrics_manager.update_metrics(
MetricsEventType::Tx,
self.update_metrics(
MetricsEventType::Transaction,
event_count as u64,
max_time_consuming_us as f64,
Some(signature),
);
Ok(())
}
fn update_metrics(
&self,
ty: MetricsEventType,
count: u64,
time_us: f64,
signature: Option<Signature>,
) {
self.metrics_manager.update_metrics(ty, count, time_us, signature);
}
}
// 实现 Clone trait 以支持模块间共享
@@ -252,8 +249,9 @@ impl Clone for EventProcessor {
parser_cache: self.parser_cache.clone(),
protocols: self.protocols.clone(),
event_type_filter: self.event_type_filter.clone(),
backpressure_strategy: self.backpressure_strategy.clone(),
backpressure_config: self.backpressure_config.clone(),
batch_config: self.batch_config.clone(),
callback: self.callback.clone(),
}
}
}
src/streaming/common/metrics.rs
+453 -242
View File
@@ -1,344 +1,554 @@
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use crossbeam::utils::Backoff;
use crossbeam::atomic::AtomicCell;
use std::sync::RwLock;
use super::config::StreamClientConfig;
use solana_sdk::signature::Signature;
use super::constants::*;
/// 单个事件类型的指标
/// 事件类型枚举
#[derive(Debug, Clone, Copy)]
pub enum EventType {
Transaction = 0,
Account = 1,
BlockMeta = 2,
}
/// 兼容性别名
pub type MetricsEventType = EventType;
impl EventType {
#[inline]
const fn as_index(self) -> usize {
self as usize
}
const fn name(self) -> &'static str {
match self {
EventType::Transaction => "TX",
EventType::Account => "Account",
EventType::BlockMeta => "Block Meta",
}
}
// 兼容性常量
pub const TX: EventType = EventType::Transaction;
}
/// 高性能原子事件指标
#[derive(Debug)]
struct AtomicEventMetrics {
process_count: AtomicU64,
events_processed: AtomicU64,
events_in_window: AtomicU64,
window_start_nanos: AtomicU64,
events_per_second_bits: AtomicU64, // f64 的位表示
}
impl AtomicEventMetrics {
fn new(now_nanos: u64) -> Self {
Self {
process_count: AtomicU64::new(0),
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),
}
}
/// 原子地增加处理计数
#[inline]
fn add_process_count(&self) {
self.process_count.fetch_add(1, Ordering::Relaxed);
}
/// 原子地增加事件处理数量
#[inline]
fn add_events_processed(&self, count: u64) {
self.events_processed.fetch_add(count, Ordering::Relaxed);
self.events_in_window.fetch_add(count, Ordering::Relaxed);
}
/// 获取当前计数（非阻塞）
#[inline]
fn get_counts(&self) -> (u64, u64, u64) {
(
self.process_count.load(Ordering::Relaxed),
self.events_processed.load(Ordering::Relaxed),
self.events_in_window.load(Ordering::Relaxed),
)
}
/// 原子地更新每秒事件数
#[inline]
fn update_events_per_second(&self, eps: f64) {
self.events_per_second_bits.store(eps.to_bits(), Ordering::Relaxed);
}
/// 获取每秒事件数
#[inline]
fn get_events_per_second(&self) -> f64 {
f64::from_bits(self.events_per_second_bits.load(Ordering::Relaxed))
}
/// 重置窗口计数
#[inline]
fn reset_window(&self, new_start_nanos: u64) {
self.events_in_window.store(0, Ordering::Relaxed);
self.window_start_nanos.store(new_start_nanos, Ordering::Relaxed);
}
#[inline]
fn get_window_start(&self) -> u64 {
self.window_start_nanos.load(Ordering::Relaxed)
}
}
/// 高性能原子处理时间统计
#[derive(Debug)]
struct AtomicProcessingTimeStats {
min_time_bits: AtomicU64,
max_time_bits: AtomicU64,
total_time_us: AtomicU64, // 存储微秒的整数部分
total_events: AtomicU64,
}
impl AtomicProcessingTimeStats {
fn new() -> Self {
Self {
min_time_bits: AtomicU64::new(f64::INFINITY.to_bits()),
max_time_bits: AtomicU64::new(0),
total_time_us: AtomicU64::new(0),
total_events: AtomicU64::new(0),
}
}
/// 原子地更新处理时间统计
#[inline]
fn update(&self, time_us: f64, event_count: u64) {
let time_bits = time_us.to_bits();
// 更新最小值（使用 compare_exchange_weak 循环）
let mut current_min = self.min_time_bits.load(Ordering::Relaxed);
while time_bits < current_min {
match self.min_time_bits.compare_exchange_weak(
current_min,
time_bits,
Ordering::Relaxed,
Ordering::Relaxed,
) {
Ok(_) => break,
Err(x) => current_min = x,
}
}
// 更新最大值
let mut current_max = self.max_time_bits.load(Ordering::Relaxed);
while time_bits > current_max {
match self.max_time_bits.compare_exchange_weak(
current_max,
time_bits,
Ordering::Relaxed,
Ordering::Relaxed,
) {
Ok(_) => break,
Err(x) => current_max = x,
}
}
// 更新累计值（将微秒转换为整数避免浮点累加问题）
let total_time_us_int = (time_us * event_count as f64) as u64;
self.total_time_us.fetch_add(total_time_us_int, Ordering::Relaxed);
self.total_events.fetch_add(event_count, Ordering::Relaxed);
}
/// 获取统计值（非阻塞）
#[inline]
fn get_stats(&self) -> ProcessingTimeStats {
let min_bits = self.min_time_bits.load(Ordering::Relaxed);
let max_bits = self.max_time_bits.load(Ordering::Relaxed);
let total_time_us_int = self.total_time_us.load(Ordering::Relaxed);
let total_events = self.total_events.load(Ordering::Relaxed);
let min_time = f64::from_bits(min_bits);
let max_time = f64::from_bits(max_bits);
let avg_time =
if total_events > 0 { total_time_us_int as f64 / total_events as f64 } else { 0.0 };
ProcessingTimeStats {
min_us: if min_time == f64::INFINITY { 0.0 } else { min_time },
max_us: max_time,
avg_us: avg_time,
}
}
}
/// 处理时间统计结果
#[derive(Debug, Clone)]
pub struct EventMetrics {
pub struct ProcessingTimeStats {
pub min_us: f64,
pub max_us: f64,
pub avg_us: f64,
}
/// 事件指标快照
#[derive(Debug, Clone)]
pub struct EventMetricsSnapshot {
pub process_count: u64,
pub events_processed: u64,
pub events_per_second: f64,
pub events_in_window: u64,
pub window_start_time: std::time::Instant,
}
impl EventMetrics {
fn new(now: std::time::Instant) -> Self {
Self {
process_count: 0,
events_processed: 0,
events_per_second: 0.0,
events_in_window: 0,
window_start_time: now,
}
}
}
/// 通用性能监控指标
/// 兼容性结构 - 完整的性能指标
#[derive(Debug, Clone)]
pub struct PerformanceMetrics {
pub start_time: std::time::Instant,
pub event_metrics: [EventMetrics; 3], // [Tx, Account, BlockMeta]
pub average_processing_time_us: f64,
pub min_processing_time_us: f64,
pub max_processing_time_us: f64,
pub last_update_time: std::time::Instant,
}
impl Default for PerformanceMetrics {
fn default() -> Self {
Self::new()
}
}
pub enum MetricsEventType {
Tx,
Account,
BlockMeta,
}
impl MetricsEventType {
fn as_index(&self) -> usize {
match self {
MetricsEventType::Tx => 0,
MetricsEventType::Account => 1,
MetricsEventType::BlockMeta => 2,
}
}
pub uptime: std::time::Duration,
pub tx_metrics: EventMetricsSnapshot,
pub account_metrics: EventMetricsSnapshot,
pub block_meta_metrics: EventMetricsSnapshot,
pub processing_stats: ProcessingTimeStats,
}
impl PerformanceMetrics {
/// 创建默认的性能指标（兼容性方法）
pub fn new() -> Self {
let now = std::time::Instant::now();
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 };
Self {
start_time: now,
event_metrics: [EventMetrics::new(now), EventMetrics::new(now), EventMetrics::new(now)],
average_processing_time_us: 0.0,
min_processing_time_us: 0.0,
max_processing_time_us: 0.0,
last_update_time: now,
uptime: std::time::Duration::ZERO,
tx_metrics: default_metrics.clone(),
account_metrics: default_metrics.clone(),
block_meta_metrics: default_metrics,
processing_stats: default_stats,
}
}
}
/// 高性能指标系统
#[derive(Debug)]
pub struct HighPerformanceMetrics {
start_nanos: u64,
event_metrics: [AtomicEventMetrics; 3],
processing_stats: AtomicProcessingTimeStats,
}
impl HighPerformanceMetrics {
fn new() -> Self {
let now_nanos =
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()
as u64;
Self {
start_nanos: now_nanos,
event_metrics: [
AtomicEventMetrics::new(now_nanos),
AtomicEventMetrics::new(now_nanos),
AtomicEventMetrics::new(now_nanos),
],
processing_stats: AtomicProcessingTimeStats::new(),
}
}
/// 更新时间窗口指标
fn update_window_metrics(
&mut self,
event_type: &MetricsEventType,
now: std::time::Instant,
window_duration: std::time::Duration,
) {
/// 获取运行时长（秒）
#[inline]
pub fn get_uptime_seconds(&self) -> f64 {
let now_nanos =
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()
as u64;
(now_nanos - self.start_nanos) as f64 / 1_000_000_000.0
}
/// 获取事件指标快照
#[inline]
pub fn get_event_metrics(&self, event_type: EventType) -> EventMetricsSnapshot {
let index = event_type.as_index();
let event_metric = &mut self.event_metrics[index];
let (process_count, events_processed, _) = self.event_metrics[index].get_counts();
let events_per_second = self.calculate_real_time_eps(event_type);
if now.duration_since(event_metric.window_start_time) >= window_duration {
let window_seconds = now.duration_since(event_metric.window_start_time).as_secs_f64();
// 修复：正确计算每秒事件数，避免除零错误
event_metric.events_per_second = if window_seconds > 0.001 {
// 避免极小的时间差
event_metric.events_in_window as f64 / window_seconds
} else {
0.0 // 时间太短时设为0，而不是事件总数
};
// 重置窗口
event_metric.events_in_window = 0;
event_metric.window_start_time = now;
}
EventMetricsSnapshot { process_count, events_processed, events_per_second }
}
/// 计算实时每秒事件数（用于显示）
fn calculate_real_time_events_per_second(
&self,
event_type: &MetricsEventType,
now: std::time::Instant,
) -> f64 {
/// 获取处理时间统计
#[inline]
pub fn get_processing_stats(&self) -> ProcessingTimeStats {
self.processing_stats.get_stats()
}
/// 计算实时每秒事件数（非阻塞）
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 current_window_duration =
now.duration_since(event_metric.window_start_time).as_secs_f64();
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);
// 如果当前窗口有足够的时间和事件，使用当前窗口的数据
if current_window_duration > 1.0 && event_metric.events_in_window > 0 {
event_metric.events_in_window as f64 / current_window_duration
// 优先级1: 当前窗口实时数据（≥2秒且有事件）
if current_window_duration_secs >= 2.0 && events_in_window > 0 {
return events_in_window as f64 / current_window_duration_secs;
}
// 如果当前窗口时间太短或没有事件，使用上一个完整窗口的值
else if event_metric.events_per_second > 0.0 {
event_metric.events_per_second
// 优先级2: 上一个窗口的结果
let stored_eps = event_metric.get_events_per_second();
if stored_eps > 0.0 {
return stored_eps;
}
// 如果都没有，计算总体平均值
else {
let total_duration = now.duration_since(self.start_time).as_secs_f64();
if total_duration > 1.0 && event_metric.events_processed > 0 {
event_metric.events_processed as f64 / total_duration
} else {
0.0
// 优先级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 =
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();
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);
}
}
}
/// 通用性能监控管理器
/// 高性能指标管理器
pub struct MetricsManager {
metrics: Arc<RwLock<PerformanceMetrics>>,
config: Arc<StreamClientConfig>,
metrics: Arc<HighPerformanceMetrics>,
enable_metrics: bool,
stream_name: String,
background_task_running: AtomicBool,
}
impl MetricsManager {
/// 创建新的性能监控管理器
pub fn new(
metrics: Arc<RwLock<PerformanceMetrics>>,
config: Arc<StreamClientConfig>,
stream_name: String,
) -> Self {
Self { metrics, config, stream_name }
/// 创建新的指标管理器
pub fn new(enable_metrics: bool, stream_name: String) -> Self {
let manager = Self {
metrics: Arc::new(HighPerformanceMetrics::new()),
enable_metrics,
stream_name,
background_task_running: AtomicBool::new(false),
};
// 启动后台任务
manager.start_background_tasks();
manager
}
/// 获取性能指标
pub fn get_metrics(&self) -> PerformanceMetrics {
// 使用 Backoff 策略进行读取尝试
let backoff = Backoff::new();
loop {
match self.metrics.read() {
Ok(metrics) => return metrics.clone(),
Err(_) => {
// 如果获取读锁失败，使用指数退避策略
backoff.snooze();
continue;
}
/// 启动后台任务
fn start_background_tasks(&self) {
if self
.background_task_running
.compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
.is_ok()
{
if !self.enable_metrics {
return;
}
let metrics = self.metrics.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_millis(500));
loop {
interval.tick().await;
let window_duration_nanos = DEFAULT_METRICS_WINDOW_SECONDS * 1_000_000_000;
// 更新所有事件类型的窗口指标
metrics.update_window_metrics(EventType::Transaction, window_duration_nanos);
metrics.update_window_metrics(EventType::Account, window_duration_nanos);
metrics.update_window_metrics(EventType::BlockMeta, window_duration_nanos);
}
});
}
}
/// 打印性能指标
pub fn print_metrics(&self) {
let metrics = self.get_metrics();
let event_names = ["TX", "Account", "Block Meta"];
let event_types =
[MetricsEventType::Tx, MetricsEventType::Account, MetricsEventType::BlockMeta];
let now = std::time::Instant::now();
/// 记录处理次数（非阻塞）
#[inline]
pub fn record_process(&self, event_type: EventType) {
if self.enable_metrics {
self.metrics.event_metrics[event_type.as_index()].add_process_count();
}
}
/// 记录事件处理（非阻塞）
#[inline]
pub fn record_events(&self, event_type: EventType, count: u64, processing_time_us: f64) {
if !self.enable_metrics {
return;
}
// 原子更新事件计数
self.metrics.event_metrics[event_type.as_index()].add_events_processed(count);
// 原子更新处理时间统计
self.metrics.processing_stats.update(processing_time_us, count);
}
/// 记录慢处理操作
#[inline]
pub fn log_slow_processing(
&self,
processing_time_us: f64,
event_count: usize,
signature: Option<Signature>,
) {
if processing_time_us > SLOW_PROCESSING_THRESHOLD_US {
log::warn!(
"{} slow processing: {:.2}us for {} events, signature: {:?}",
self.stream_name,
processing_time_us,
event_count,
signature
);
}
}
/// 获取运行时长
pub fn get_uptime(&self) -> std::time::Duration {
std::time::Duration::from_secs_f64(self.metrics.get_uptime_seconds())
}
/// 获取事件指标
pub fn get_event_metrics(&self, event_type: EventType) -> EventMetricsSnapshot {
self.metrics.get_event_metrics(event_type)
}
/// 获取处理时间统计
pub fn get_processing_stats(&self) -> ProcessingTimeStats {
self.metrics.get_processing_stats()
}
/// 打印性能指标（非阻塞）
pub fn print_metrics(&self) {
println!("\n📊 {} Performance Metrics", self.stream_name);
println!(" Run Time: {:?}", metrics.start_time.elapsed());
// 打印表格头部
println!(" Run Time: {:?}", self.get_uptime());
// 打印事件指标表格
println!("┌─────────────┬──────────────┬──────────────────┬─────────────────┐");
println!("│ Event Type │ Process Count│ Events Processed │ Events/Second │");
println!("├─────────────┼──────────────┼──────────────────┼─────────────────┤");
// 打印每种事件类型的数据
for (i, name) in event_names.iter().enumerate() {
let event_metric = &metrics.event_metrics[i];
// 使用实时计算的每秒事件数，而不是窗口更新的值
let real_time_eps = metrics.calculate_real_time_events_per_second(&event_types[i], now);
for event_type in [EventType::Transaction, EventType::Account, EventType::BlockMeta] {
let metrics = self.get_event_metrics(event_type);
println!(
"│ {:11} │ {:12} │ {:16} │ {:13.2} │",
name,
event_metric.process_count,
event_metric.events_processed,
real_time_eps
event_type.name(),
metrics.process_count,
metrics.events_processed,
metrics.events_per_second
);
}
println!("└─────────────┴──────────────┴──────────────────┴─────────────────┘");
// 打印处理时间统计表格
let stats = self.get_processing_stats();
println!("\n⏱️ Processing Time Statistics");
println!("┌─────────────────────┬─────────────┐");
println!("│ Metric │ Value (us) │");
println!("├─────────────────────┼─────────────┤");
println!("│ Average │ {:9.2} │", metrics.average_processing_time_us);
println!("│ Minimum │ {:9.2} │", metrics.min_processing_time_us);
println!("│ Maximum │ {:9.2} │", metrics.max_processing_time_us);
println!("│ Average │ {:9.2} │", stats.avg_us);
println!("│ Minimum │ {:9.2} │", stats.min_us);
println!("│ Maximum │ {:9.2} │", stats.max_us);
println!("└─────────────────────┴─────────────┘");
println!();
}
/// 启动自动性能监控任务
pub async fn start_auto_monitoring(&self) -> Option<tokio::task::JoinHandle<()>> {
// 检查是否启用性能监控
if !self.config.enable_metrics {
return None; // 如果未启用性能监控，不启动监控任务
if !self.enable_metrics {
return None;
}
let metrics_manager = self.clone();
let manager = self.clone();
let handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(
let mut interval = tokio::time::interval(std::time::Duration::from_secs(
DEFAULT_METRICS_PRINT_INTERVAL_SECONDS,
));
loop {
interval.tick().await;
metrics_manager.print_metrics();
manager.print_metrics();
}
});
Some(handle)
}
/// 更新处理次数
pub fn add_process_count(&self, event_type: MetricsEventType) {
if !self.config.enable_metrics {
return;
}
// 使用 Backoff 策略进行写入尝试
let backoff = Backoff::new();
loop {
match self.metrics.write() {
Ok(mut metrics) => {
metrics.event_metrics[event_type.as_index()].process_count += 1;
break;
},
Err(_) => {
// 如果获取写锁失败，使用指数退避策略
backoff.snooze();
continue;
}
}
// === 兼容性方法 ===
/// 兼容性构造函数
pub fn new_with_metrics(
_metrics: Arc<std::sync::RwLock<PerformanceMetrics>>,
enable_metrics: bool,
stream_name: String,
) -> Self {
Self::new(enable_metrics, stream_name)
}
/// 获取完整的性能指标（兼容性方法）
pub fn get_metrics(&self) -> PerformanceMetrics {
PerformanceMetrics {
uptime: self.get_uptime(),
tx_metrics: self.get_event_metrics(EventType::Transaction),
account_metrics: self.get_event_metrics(EventType::Account),
block_meta_metrics: self.get_event_metrics(EventType::BlockMeta),
processing_stats: self.get_processing_stats(),
}
}
// 保持向后兼容的方法
/// 兼容性方法 - 添加交易处理计数
#[inline]
pub fn add_tx_process_count(&self) {
self.add_process_count(MetricsEventType::Tx);
self.record_process(EventType::Transaction);
}
/// 兼容性方法 - 添加账户处理计数
#[inline]
pub fn add_account_process_count(&self) {
self.add_process_count(MetricsEventType::Account);
self.record_process(EventType::Account);
}
/// 兼容性方法 - 添加区块元数据处理计数
#[inline]
pub fn add_block_meta_process_count(&self) {
self.add_process_count(MetricsEventType::BlockMeta);
self.record_process(EventType::BlockMeta);
}
/// 更新性能指标
/// 兼容性方法 - 更新指标
#[inline]
pub fn update_metrics(
&self,
event_type: MetricsEventType,
events_processed: u64,
processing_time_us: f64,
signature: Option<Signature>,
) {
// 检查是否启用性能监控
if !self.config.enable_metrics {
return;
}
// 使用 Backoff 策略进行写入尝试
let backoff = Backoff::new();
loop {
match self.metrics.write() {
Ok(mut metrics) => {
let now = std::time::Instant::now();
let index = event_type.as_index();
// 更新事件计数
metrics.event_metrics[index].events_processed += events_processed;
metrics.event_metrics[index].events_in_window += events_processed;
metrics.last_update_time = now;
// 更新处理时间统计
if processing_time_us < metrics.min_processing_time_us
|| metrics.min_processing_time_us == 0.0
{
metrics.min_processing_time_us = processing_time_us;
}
if processing_time_us > metrics.max_processing_time_us {
metrics.max_processing_time_us = processing_time_us;
}
// 计算平均处理时间 - 使用增量更新避免重复计算
let total_events = metrics.event_metrics[index].events_processed;
if total_events > 0 {
let total_events_f64 = total_events as f64;
let old_total = (total_events_f64 - events_processed as f64).max(0.0);
metrics.average_processing_time_us = if old_total > 0.0 {
(metrics.average_processing_time_us * old_total
+ processing_time_us * events_processed as f64)
/ total_events_f64
} else {
processing_time_us
};
}
// 更新时间窗口指标
let window_duration = std::time::Duration::from_secs(DEFAULT_METRICS_WINDOW_SECONDS);
metrics.update_window_metrics(&event_type, now, window_duration);
break;
},
Err(_) => {
// 如果获取写锁失败，使用指数退避策略
backoff.snooze();
continue;
}
}
}
}
/// 记录慢处理操作
pub fn log_slow_processing(&self, processing_time_us: f64, event_count: usize) {
if processing_time_us > SLOW_PROCESSING_THRESHOLD_US {
log::warn!(
"{} slow processing: {processing_time_us}us for {event_count} events",
self.stream_name
);
}
self.record_events(event_type, events_processed, processing_time_us);
self.log_slow_processing(processing_time_us, events_processed as usize, signature);
}
}
@@ -346,8 +556,9 @@ impl Clone for MetricsManager {
fn clone(&self) -> Self {
Self {
metrics: self.metrics.clone(),
config: self.config.clone(),
enable_metrics: self.enable_metrics,
stream_name: self.stream_name.clone(),
background_task_running: AtomicBool::new(false), // 新实例不自动启动后台任务
}
}
}
src/streaming/event_parser/common/mod.rs
+6 -3
View File
@@ -54,7 +54,7 @@ macro_rules! impl_unified_event {
Box::new(self.clone())
}
fn merge(&mut self, other: Box<dyn $crate::streaming::event_parser::core::traits::UnifiedEvent>) {
fn merge(&mut self, other: &dyn $crate::streaming::event_parser::core::traits::UnifiedEvent) {
if let Some(_e) = other.as_any().downcast_ref::<$struct_name>() {
$(
self.$field = _e.$field.clone();
@@ -66,10 +66,13 @@ macro_rules! impl_unified_event {
self.metadata.set_swap_data(swap_data);
}
fn index(&self) -> String {
self.metadata.index.clone()
fn instruction_outer_index(&self) -> i64 {
self.metadata.instruction_outer_index
}
fn instruction_inner_index(&self) -> Option<i64> {
self.metadata.instruction_inner_index
}
fn transaction_index(&self) -> Option<u64> {
self.metadata.transaction_index
}
src/streaming/event_parser/common/types.rs
+10 -51
View File
@@ -55,36 +55,9 @@ impl EventMetadataPool {
}
}
/// Transfer data object pool
pub struct TransferDataPool {
pool: Arc<ArrayQueue<TransferData>>,
}
impl Default for TransferDataPool {
fn default() -> Self {
Self::new()
}
}
impl TransferDataPool {
pub fn new() -> Self {
Self { pool: Arc::new(ArrayQueue::new(TRANSFER_DATA_POOL_SIZE)) }
}
pub fn acquire(&self) -> Option<TransferData> {
self.pool.pop()
}
pub fn release(&self, transfer_data: TransferData) {
// 如果队列已满，push 会失败，但不会阻塞
let _ = self.pool.push(transfer_data);
}
}
// Global object pool instances
lazy_static::lazy_static! {
pub static ref EVENT_METADATA_POOL: EventMetadataPool = EventMetadataPool::new();
pub static ref TRANSFER_DATA_POOL: TransferDataPool = TransferDataPool::new();
}
#[derive(
@@ -304,20 +277,6 @@ impl ProtocolInfo {
}
}
/// Transfer data
#[derive(
Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
)]
pub struct TransferData {
pub token_program: Pubkey,
pub source: Pubkey,
pub destination: Pubkey,
pub authority: Option<Pubkey>,
pub amount: u64,
pub decimals: Option<u8>,
pub mint: Option<Pubkey>,
}
#[derive(
Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
)]
@@ -337,7 +296,7 @@ pub struct EventMetadata {
pub id: String,
pub signature: String,
pub slot: u64,
pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
pub block_time: i64,
pub block_time_ms: i64,
pub program_received_time_us: i64,
@@ -345,10 +304,9 @@ pub struct EventMetadata {
pub protocol: ProtocolType,
pub event_type: EventType,
pub program_id: Pubkey,
#[deprecated(note = "Please use swap_data instead")]
pub transfer_datas: Vec<TransferData>,
pub swap_data: Option<SwapData>,
pub index: String, // 保留原有的指令索引
pub instruction_outer_index: i64,
pub instruction_inner_index: Option<i64>,
}
impl EventMetadata {
@@ -362,14 +320,15 @@ impl EventMetadata {
protocol: ProtocolType,
event_type: EventType,
program_id: Pubkey,
index: String,
instruction_outer_index: i64,
instruction_inner_index: Option<i64>,
program_received_time_us: i64,
) -> Self {
Self {
id,
signature,
slot,
transaction_index: None, // 默认为None，后续设置
transaction_index: None, // 默认为None，后续设置
block_time,
block_time_ms,
program_received_time_us,
@@ -377,9 +336,9 @@ impl EventMetadata {
protocol,
event_type,
program_id,
transfer_datas: vec![],
swap_data: None,
index,
instruction_outer_index,
instruction_inner_index,
}
}
@@ -413,7 +372,7 @@ lazy_static::lazy_static! {
/// Parse token transfer data from next instructions
pub fn parse_swap_data_from_next_instructions(
event: Box<dyn UnifiedEvent>,
event: &dyn UnifiedEvent,
inner_instruction: &solana_transaction_status::InnerInstructions,
current_index: i8,
accounts: &[Pubkey],
@@ -435,7 +394,7 @@ pub fn parse_swap_data_from_next_instructions(
let mut from_vault: Option<Pubkey> = None;
let mut to_vault: Option<Pubkey> = None;
match_event!(event, {
match_event!(&*event, {
BonkTradeEvent => |e: BonkTradeEvent| {
user = Some(e.payer);
from_mint = Some(e.base_token_mint);
src/streaming/event_parser/core/traits.rs
+160 -85
View File
@@ -7,6 +7,7 @@ use solana_sdk::{
use solana_transaction_status::{InnerInstructions, TransactionWithStatusMeta};
use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::Arc;
use crate::streaming::event_parser::common::{parse_swap_data_from_next_instructions, SwapData};
use crate::streaming::event_parser::protocols::pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent};
@@ -17,7 +18,6 @@ use crate::streaming::event_parser::{
pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent},
},
};
use crate::streaming::shred::MetricsEventType;
/// Unified Event Interface - All protocol events must implement this trait
pub trait UnifiedEvent: Debug + Send + Sync {
@@ -55,7 +55,7 @@ pub trait UnifiedEvent: Debug + Send + Sync {
fn clone_boxed(&self) -> Box<dyn UnifiedEvent>;
/// Merge events (optional implementation)
fn merge(&mut self, _other: Box<dyn UnifiedEvent>) {
fn merge(&mut self, _other: &dyn UnifiedEvent) {
// Default implementation: no merging operation
}
@@ -63,7 +63,8 @@ pub trait UnifiedEvent: Debug + Send + Sync {
fn set_swap_data(&mut self, swap_data: SwapData);
/// Get index
fn index(&self) -> String;
fn instruction_outer_index(&self) -> i64;
fn instruction_inner_index(&self) -> Option<i64>;
/// Get transaction index in slot
fn transaction_index(&self) -> Option<u64>;
@@ -88,7 +89,8 @@ pub trait EventParser: Send + Sync {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>>;
/// 从指令中解析事件数据
@@ -101,7 +103,8 @@ pub trait EventParser: Send + Sync {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>>;
/// 从VersionedTransaction中解析指令事件的通用方法
@@ -144,7 +147,8 @@ pub trait EventParser: Send + Sync {
slot,
block_time,
program_received_time_us,
format!("{index}"),
index as i64,
None,
)
.await
{
@@ -156,7 +160,7 @@ pub trait EventParser: Send + Sync {
{
events.iter_mut().for_each(|event| {
let swap_data = parse_swap_data_from_next_instructions(
event.clone_boxed(),
event.as_ref(),
inn,
-1_i8,
&accounts,
@@ -217,62 +221,100 @@ pub trait EventParser: Send + Sync {
let mut inner_instructions: Vec<InnerInstructions> = vec![];
if let Some(meta) = meta {
inner_instructions = meta.inner_instructions.unwrap_or_default();
for loopup in meta.loaded_addresses.writable {
address_table_lookups.push(loopup);
}
for loopup in meta.loaded_addresses.readonly {
address_table_lookups.push(loopup);
}
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<Pubkey> = vec![];
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);
// 预分配容量，避免动态扩容
let mut instruction_events: Vec<Box<dyn UnifiedEvent>> = Vec::with_capacity(16);
let mut inner_instruction_events: Vec<Box<dyn UnifiedEvent>> = Vec::with_capacity(8);
// 解析指令事件
accounts = versioned_tx.message.static_account_keys().to_vec();
accounts.extend(address_table_lookups.clone());
instruction_events = self
.parse_instruction_events_from_versioned_transaction(
let accounts_for_task1 = Arc::clone(&accounts_arc);
let inner_instructions_for_task1 = Arc::clone(&inner_instructions_arc);
let task1 = async move {
self.parse_instruction_events_from_versioned_transaction(
&versioned_tx,
signature,
slot,
block_time,
program_received_time_us,
&accounts,
&inner_instructions,
&accounts_for_task1,
&inner_instructions_for_task1,
)
.await
.unwrap_or_else(|_e| vec![]);
.unwrap_or_else(|_e| vec![])
};
// 解析内联指令事件
// 预分配容量，避免动态扩容
let mut inner_instruction_events: Vec<Box<dyn UnifiedEvent>> = Vec::with_capacity(8);
// 检查交易是否成功
for inner_instruction in inner_instructions {
let inner_instructions_for_task2 = Arc::clone(&inner_instructions_arc);
let accounts_for_task2_1 = Arc::clone(&accounts_arc);
let accounts_for_task2_2 = Arc::clone(&accounts_arc);
let mut task2_params = Vec::with_capacity(inner_instructions_for_task2.len() * 5);
for inner_instruction in inner_instructions_for_task2.iter() {
for (index, instruction) in inner_instruction.instructions.iter().enumerate() {
// 解析嵌套指令
let compiled_instruction = instruction.instruction.clone();
task2_params.push((
&instruction.instruction,
signature,
slot,
block_time,
program_received_time_us,
inner_instruction.index as i64,
Some(index as i64),
inner_instruction,
));
}
}
// 转换为 Arc<[T]> 更轻量
let task2_params: Arc<[_]> = task2_params.into();
let task2_params_clone = Arc::clone(&task2_params);
let task2_1 = async move {
let mut instruction_events: Vec<Box<dyn UnifiedEvent>> = Vec::with_capacity(16);
for (
instruction,
signature,
slot,
block_time,
program_received_time_us,
outer_index,
inner_index,
inner_instruction,
) in task2_params_clone.iter()
{
if let Ok(mut events) = self
.parse_instruction(
&compiled_instruction,
&accounts,
signature,
slot,
block_time,
program_received_time_us,
format!("{}.{}", inner_instruction.index, index),
instruction,
&accounts_for_task2_1,
*signature,
*slot,
*block_time,
*program_received_time_us,
*outer_index,
*inner_index,
)
.await
{
if !events.is_empty() {
events.iter_mut().for_each(|event| {
let swap_data = parse_swap_data_from_next_instructions(
event.clone_boxed(),
event.as_ref(),
&inner_instruction,
index as i8,
&accounts,
inner_index.unwrap_or_default() as i8,
&accounts_for_task2_1,
);
if let Some(swap_data) = swap_data {
event.set_swap_data(swap_data);
@@ -281,24 +323,41 @@ pub trait EventParser: Send + Sync {
instruction_events.extend(events);
}
}
}
instruction_events
};
let task2_2 = async move {
let mut inner_instruction_events: Vec<Box<dyn UnifiedEvent>> = Vec::with_capacity(8);
for (
instruction,
signature,
slot,
block_time,
program_received_time_us,
outer_index,
inner_index,
inner_instruction,
) in task2_params.iter()
{
if let Ok(mut events) = self
.parse_inner_instruction(
&compiled_instruction,
signature,
slot,
block_time,
program_received_time_us,
format!("{}.{}", inner_instruction.index, index),
instruction,
*signature,
*slot,
*block_time,
*program_received_time_us,
*outer_index,
*inner_index,
)
.await
{
if !events.is_empty() {
events.iter_mut().for_each(|event| {
let swap_data = parse_swap_data_from_next_instructions(
event.clone_boxed(),
event.as_ref(),
&inner_instruction,
index as i8,
&accounts,
inner_index.unwrap_or_default() as i8,
&accounts_for_task2_2,
);
if let Some(swap_data) = swap_data {
event.set_swap_data(swap_data);
@@ -308,39 +367,41 @@ pub trait EventParser: Send + Sync {
}
}
}
}
inner_instruction_events
};
let (r1, r2_1, r2_2) = tokio::join!(task1, task2_1, task2_2);
instruction_events.extend(r1);
instruction_events.extend(r2_1);
inner_instruction_events.extend(r2_2);
if !instruction_events.is_empty() && !inner_instruction_events.is_empty() {
for instruction_event in &mut instruction_events {
for inner_instruction_event in &inner_instruction_events {
if instruction_event.id() == inner_instruction_event.id() {
let i_index = instruction_event.index();
let in_index = inner_instruction_event.index();
if !i_index.contains(".") && in_index.contains(".") {
let in_index_parts: Vec<&str> = in_index.split(".").collect();
if !in_index_parts.is_empty() && in_index_parts[0] == i_index {
instruction_event.merge(inner_instruction_event.clone_boxed());
if instruction_event.instruction_inner_index().is_none()
&& inner_instruction_event.instruction_inner_index().is_some()
{
if inner_instruction_event.instruction_outer_index()
== instruction_event.instruction_outer_index()
{
instruction_event.merge(inner_instruction_event.as_ref());
break;
}
} else if i_index.contains(".") && in_index.contains(".") {
// 嵌套指令
let i_index_parts: Vec<&str> = i_index.split(".").collect();
let in_index_parts: Vec<&str> = in_index.split(".").collect();
if !i_index_parts.is_empty()
&& !in_index_parts.is_empty()
&& i_index_parts[0] == in_index_parts[0]
} else if instruction_event.instruction_inner_index().is_some()
&& inner_instruction_event.instruction_inner_index().is_some()
{
if instruction_event.instruction_outer_index()
== inner_instruction_event.instruction_outer_index()
{
let i_index_child_index = i_index_parts
.get(1)
.and_then(|s| s.parse::<u32>().ok())
.unwrap_or(0);
let in_index_child_index = in_index_parts
.get(1)
.and_then(|s| s.parse::<u32>().ok())
.unwrap_or(0);
if in_index_child_index > i_index_child_index {
instruction_event.merge(inner_instruction_event.clone_boxed());
if inner_instruction_event
.instruction_inner_index()
.unwrap_or_default()
> instruction_event
.instruction_inner_index()
.unwrap_or_default()
{
instruction_event.merge(inner_instruction_event.as_ref());
break;
}
}
@@ -433,7 +494,8 @@ pub trait EventParser: Send + Sync {
slot: Option<u64>,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Result<Vec<Box<dyn UnifiedEvent>>> {
let slot = slot.unwrap_or(0);
let events = self.parse_events_from_inner_instruction(
@@ -442,7 +504,8 @@ pub trait EventParser: Send + Sync {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
);
Ok(events)
}
@@ -456,7 +519,8 @@ pub trait EventParser: Send + Sync {
slot: Option<u64>,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Result<Vec<Box<dyn UnifiedEvent>>> {
let slot = slot.unwrap_or(0);
let events = self.parse_events_from_instruction(
@@ -466,7 +530,8 @@ pub trait EventParser: Send + Sync {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
);
Ok(events)
}
@@ -543,7 +608,8 @@ impl GenericEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Option<Box<dyn UnifiedEvent>> {
if let Some(parser) = config.inner_instruction_parser {
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
@@ -557,7 +623,8 @@ impl GenericEventParser {
config.protocol_type.clone(),
config.event_type.clone(),
config.program_id,
index,
outer_index,
inner_index,
program_received_time_us,
);
parser(data, metadata)
@@ -577,7 +644,8 @@ impl GenericEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Option<Box<dyn UnifiedEvent>> {
if let Some(parser) = config.instruction_parser {
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
@@ -591,7 +659,8 @@ impl GenericEventParser {
config.protocol_type.clone(),
config.event_type.clone(),
config.program_id,
index,
outer_index,
inner_index,
program_received_time_us,
);
parser(data, account_pubkeys, metadata)
@@ -604,9 +673,11 @@ impl GenericEventParser {
#[async_trait::async_trait]
impl EventParser for GenericEventParser {
fn inner_instruction_configs(&self) -> HashMap<&'static str, Vec<GenericEventParseConfig>> {
// 返回引用而非克隆，减少内存分配
self.inner_instruction_configs.clone()
}
fn instruction_configs(&self) -> HashMap<Vec<u8>, Vec<GenericEventParseConfig>> {
// 返回引用而非克隆，减少内存分配
self.instruction_configs.clone()
}
/// 从内联指令中解析事件数据
@@ -618,7 +689,8 @@ impl EventParser for GenericEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
let inner_instruction_data_decoded = inner_instruction.data.clone();
if inner_instruction_data_decoded.len() < 16 {
@@ -638,7 +710,8 @@ impl EventParser for GenericEventParser {
slot,
block_time,
program_received_time_us,
index.clone(),
outer_index,
inner_index,
) {
events.push(event);
}
@@ -658,7 +731,8 @@ impl EventParser for GenericEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
let program_id = accounts[instruction.program_id_index as usize];
if !self.should_handle(&program_id) {
@@ -691,7 +765,8 @@ impl EventParser for GenericEventParser {
slot,
block_time,
program_received_time_us,
index.clone(),
outer_index,
inner_index,
) {
events.push(event);
}
src/streaming/event_parser/protocols/block/block_meta_event.rs
+2 -1
View File
@@ -28,7 +28,8 @@ impl BlockMetaEvent {
crate::streaming::event_parser::common::types::ProtocolType::Common,
EventType::BlockMeta,
solana_sdk::pubkey::Pubkey::default(),
"".to_string(),
0,
None,
program_received_time_us,
);
Self { metadata, slot, block_hash }
src/streaming/event_parser/protocols/bonk/parser.rs
+8 -4
View File
@@ -618,7 +618,8 @@ impl EventParser for BonkEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_inner_instruction(
inner_instruction,
@@ -626,7 +627,8 @@ impl EventParser for BonkEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
@@ -638,7 +640,8 @@ impl EventParser for BonkEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_instruction(
instruction,
@@ -647,7 +650,8 @@ impl EventParser for BonkEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
src/streaming/event_parser/protocols/mutil/parser.rs
+8 -4
View File
@@ -79,7 +79,8 @@ impl EventParser for MutilEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_inner_instruction(
inner_instruction,
@@ -87,7 +88,8 @@ impl EventParser for MutilEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
@@ -99,7 +101,8 @@ impl EventParser for MutilEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_instruction(
instruction,
@@ -108,7 +111,8 @@ impl EventParser for MutilEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
src/streaming/event_parser/protocols/pumpfun/parser.rs
+8 -4
View File
@@ -299,7 +299,8 @@ impl EventParser for PumpFunEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_inner_instruction(
inner_instruction,
@@ -307,7 +308,8 @@ impl EventParser for PumpFunEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
@@ -319,7 +321,8 @@ impl EventParser for PumpFunEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_instruction(
instruction,
@@ -328,7 +331,8 @@ impl EventParser for PumpFunEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
src/streaming/event_parser/protocols/pumpswap/parser.rs
+8 -4
View File
@@ -389,7 +389,8 @@ impl EventParser for PumpSwapEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_inner_instruction(
inner_instruction,
@@ -397,7 +398,8 @@ impl EventParser for PumpSwapEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
@@ -409,7 +411,8 @@ impl EventParser for PumpSwapEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_instruction(
instruction,
@@ -418,7 +421,8 @@ impl EventParser for PumpSwapEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs
+8 -4
View File
@@ -391,7 +391,8 @@ impl EventParser for RaydiumAmmV4EventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_inner_instruction(
inner_instruction,
@@ -399,7 +400,8 @@ impl EventParser for RaydiumAmmV4EventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
@@ -411,7 +413,8 @@ impl EventParser for RaydiumAmmV4EventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_instruction(
instruction,
@@ -420,7 +423,8 @@ impl EventParser for RaydiumAmmV4EventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
src/streaming/event_parser/protocols/raydium_clmm/parser.rs
+8 -4
View File
@@ -432,7 +432,8 @@ impl EventParser for RaydiumClmmEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_inner_instruction(
inner_instruction,
@@ -440,7 +441,8 @@ impl EventParser for RaydiumClmmEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
@@ -452,7 +454,8 @@ impl EventParser for RaydiumClmmEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_instruction(
instruction,
@@ -461,7 +464,8 @@ impl EventParser for RaydiumClmmEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
src/streaming/event_parser/protocols/raydium_cpmm/parser.rs
+8 -4
View File
@@ -282,7 +282,8 @@ impl EventParser for RaydiumCpmmEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_inner_instruction(
inner_instruction,
@@ -290,7 +291,8 @@ impl EventParser for RaydiumCpmmEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
@@ -302,7 +304,8 @@ impl EventParser for RaydiumCpmmEventParser {
slot: u64,
block_time: Option<Timestamp>,
program_received_time_us: i64,
index: String,
outer_index: i64,
inner_index: Option<i64>,
) -> Vec<Box<dyn UnifiedEvent>> {
self.inner.parse_events_from_instruction(
instruction,
@@ -311,7 +314,8 @@ impl EventParser for RaydiumCpmmEventParser {
slot,
block_time,
program_received_time_us,
index,
outer_index,
inner_index,
)
}
src/streaming/grpc/stream_handler.rs
+23 -64
View File
@@ -8,7 +8,6 @@ use yellowstone_grpc_proto::geyser::{
use super::types::{BlockMetaPretty, EventPretty, TransactionPretty};
use crate::common::AnyResult;
use crate::streaming::common::EventProcessor;
use crate::streaming::event_parser::UnifiedEvent;
use crate::streaming::grpc::AccountPretty;
/// 流消息处理器
@@ -16,16 +15,12 @@ pub struct StreamHandler;
impl StreamHandler {
/// 处理单个流消息
pub async fn handle_stream_message<F>(
pub async fn handle_stream_message(
msg: SubscribeUpdate,
subscribe_tx: &mut (impl Sink<SubscribeRequest, Error = mpsc::SendError> + Unpin),
event_processor: EventProcessor,
callback: &F,
bot_wallet: Option<Pubkey>,
) -> AnyResult<()>
where
F: Fn(Box<dyn UnifiedEvent>) + Send + Sync,
{
) -> AnyResult<()> {
let created_at = msg.created_at;
match msg.update_oneof {
Some(UpdateOneof::Account(account)) => {
@@ -34,7 +29,6 @@ impl StreamHandler {
event_processor
.process_grpc_event_transaction_with_metrics(
EventPretty::Account(account_pretty),
callback,
bot_wallet,
)
.await?;
@@ -45,7 +39,6 @@ impl StreamHandler {
event_processor
.process_grpc_event_transaction_with_metrics(
EventPretty::BlockMeta(block_meta_pretty),
callback,
bot_wallet,
)
.await?;
@@ -60,7 +53,6 @@ impl StreamHandler {
event_processor
.process_grpc_event_transaction_with_metrics(
EventPretty::Transaction(transaction_pretty),
callback,
bot_wallet,
)
.await?;
@@ -84,58 +76,25 @@ impl StreamHandler {
Ok(())
}
// /// 处理背压策略
// async fn handle_backpressure(
// tx: &mut mpsc::Sender<EventPretty>,
// event_pretty: EventPretty,
// backpressure_strategy: BackpressureStrategy,
// ) -> AnyResult<()> {
// match backpressure_strategy {
// BackpressureStrategy::Block => {
// // 阻塞等待，直到有空间
// if let Err(e) = tx.send(event_pretty).await {
// log::error!("Failed to send transaction to channel: {:?}", e);
// return Err(anyhow::anyhow!("Channel send failed: {:?}", e));
// }
// }
// BackpressureStrategy::Drop => {
// // 尝试发送，如果失败则丢弃
// if let Err(e) = tx.try_send(event_pretty) {
// if e.is_full() {
// log::warn!("Channel is full, dropping transaction");
// } else {
// log::error!("Channel is closed: {:?}", e);
// return Err(anyhow::anyhow!("Channel is closed: {:?}", e));
// }
// }
// }
// BackpressureStrategy::Retry { max_attempts, wait_ms } => {
// // 重试有限次数
// let mut retry_count = 0;
// loop {
// match tx.try_send(event_pretty.clone()) {
// Ok(_) => break,
// Err(e) => {
// if e.is_full() {
// retry_count += 1;
// if retry_count >= max_attempts {
// log::warn!(
// "Channel is full after {} attempts, dropping transaction",
// retry_count
// );
// break;
// }
// tokio::time::sleep(tokio::time::Duration::from_millis(wait_ms))
// .await;
// } else {
// log::error!("Channel is closed: {:?}", e);
// return Err(anyhow::anyhow!("Channel is closed: {:?}", e));
// }
// }
// }
// }
// }
// }
// Ok(())
// }
pub async fn handle_stream_system_message(
msg: SubscribeUpdate,
subscribe_tx: &mut (impl Sink<SubscribeRequest, Error = mpsc::SendError> + Unpin),
) -> AnyResult<Option<EventPretty>> {
let created_at = msg.created_at;
let event_pretty = match msg.update_oneof {
Some(UpdateOneof::Transaction(sut)) => Some(TransactionPretty::from((sut, created_at))),
Some(UpdateOneof::Ping(_)) => {
subscribe_tx
.send(SubscribeRequest {
ping: Some(SubscribeRequestPing { id: 1 }),
..Default::default()
})
.await?;
None
}
Some(UpdateOneof::Pong(_)) => None,
_ => None,
};
Ok(event_pretty.map(|e| EventPretty::Transaction(e)))
}
}
src/streaming/grpc/types.rs
+7 -6
View File
@@ -69,7 +69,7 @@ impl fmt::Debug for BlockMetaPretty {
#[derive(Clone)]
pub struct TransactionPretty {
pub slot: u64,
pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
pub block_hash: String,
pub block_time: Option<Timestamp>,
pub signature: Signature,
@@ -95,10 +95,11 @@ impl From<SubscribeUpdateAccount> for AccountPretty {
let account_info = account.account.unwrap();
Self {
slot: account.slot,
signature: Signature::try_from(
account_info.txn_signature.unwrap_or_default().as_slice(),
)
.expect("valid signature"),
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,
@@ -138,7 +139,7 @@ impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty
let transaction_index = tx.index;
Self {
slot,
transaction_index: Some(transaction_index), // 提取交易索引
transaction_index: Some(transaction_index), // 提取交易索引
block_time,
block_hash: "".to_string(),
signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
src/streaming/mod.rs
+2 -2
View File
@@ -4,8 +4,8 @@ pub mod grpc;
pub mod shred;
pub mod shred_stream;
pub mod yellowstone_grpc;
// pub mod yellowstone_sub_system;
pub mod yellowstone_sub_system;
pub use shred::ShredStreamGrpc;
pub use yellowstone_grpc::YellowstoneGrpc;
// pub use yellowstone_sub_system::{SystemEvent, TransferInfo};
pub use yellowstone_sub_system::{SystemEvent, TransferInfo};
src/streaming/shred/connection.rs
+1 -3
View File
@@ -29,10 +29,8 @@ impl ShredStreamGrpc {
pub async fn new_with_config(endpoint: String, config: StreamClientConfig) -> AnyResult<Self> {
let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?;
let metrics = Arc::new(RwLock::new(PerformanceMetrics::new()));
let config_arc = Arc::new(config.clone());
let metrics_manager =
MetricsManager::new(metrics.clone(), config_arc, "ShredStream".to_string());
let metrics_manager = MetricsManager::new(config.enable_metrics, "ShredStream".to_string());
Ok(Self {
shredstream_client: Arc::new(shredstream_client),
src/streaming/shred_stream.rs
+5 -7
View File
@@ -1,3 +1,5 @@
use std::sync::Arc;
use futures::StreamExt;
use solana_sdk::pubkey::Pubkey;
@@ -39,8 +41,9 @@ impl ShredStreamGrpc {
event_processor.set_protocols_and_event_type_filter(
protocols,
event_type_filter,
self.config.backpressure.strategy,
self.config.backpressure.clone(),
self.config.batch.clone(),
Some(Arc::new(callback)),
);
// 启动流处理
@@ -61,7 +64,6 @@ impl ShredStreamGrpc {
.process_shred_transaction_immediate(
transaction_with_slot,
bot_wallet,
&callback,
)
.await
{
@@ -82,11 +84,7 @@ impl ShredStreamGrpc {
});
// 保存订阅句柄
let subscription_handle = SubscriptionHandle::new(
stream_task,
event_processor.get_event_handle(),
metrics_handle,
);
let subscription_handle = SubscriptionHandle::new(stream_task, None, metrics_handle);
let mut handle_guard = self.subscription_handle.lock().await;
*handle_guard = Some(subscription_handle);
src/streaming/yellowstone_grpc.rs
+8 -10
View File
@@ -51,12 +51,14 @@ impl YellowstoneGrpc {
) -> AnyResult<Self> {
let _ = rustls::crypto::ring::default_provider().install_default().ok();
let metrics = Arc::new(RwLock::new(PerformanceMetrics::new()));
let config_arc = Arc::new(config.clone());
let subscription_manager =
SubscriptionManager::new(endpoint.clone(), x_token.clone(), config.clone());
let metrics_manager =
MetricsManager::new(metrics.clone(), config_arc.clone(), "YellowstoneGrpc".to_string());
let metrics_manager = MetricsManager::new_with_metrics(
metrics.clone(),
config.enable_metrics,
"YellowstoneGrpc".to_string(),
);
let event_processor = EventProcessor::new(metrics_manager.clone(), config.clone());
Ok(Self {
@@ -179,8 +181,9 @@ impl YellowstoneGrpc {
event_processor.set_protocols_and_event_type_filter(
protocols,
event_type_filter,
self.config.backpressure.strategy,
self.config.backpressure.clone(),
self.config.batch.clone(),
Some(Arc::new(callback)),
);
let stream_handle = tokio::spawn(async move {
while let Some(message) = stream.next().await {
@@ -190,7 +193,6 @@ impl YellowstoneGrpc {
msg,
&mut subscribe_tx,
event_processor.clone(),
&callback,
bot_wallet,
)
.await
@@ -208,11 +210,7 @@ impl YellowstoneGrpc {
});
// 保存订阅句柄
let subscription_handle = SubscriptionHandle::new(
stream_handle,
self.event_processor.get_event_handle(),
metrics_handle,
);
let subscription_handle = SubscriptionHandle::new(stream_handle, None, metrics_handle);
let mut handle_guard = self.subscription_handle.lock().await;
*handle_guard = Some(subscription_handle);
src/streaming/yellowstone_sub_system.rs
+12 -20
View File
@@ -1,19 +1,17 @@
use crate::{
common::AnyResult,
streaming::{
grpc::{BackpressureStrategy, EventPretty, StreamHandler},
grpc::{EventPretty, StreamHandler},
yellowstone_grpc::YellowstoneGrpc,
},
};
use futures::{channel::mpsc, StreamExt};
use futures::StreamExt;
use log::error;
use solana_program::pubkey;
use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction};
use solana_transaction_status::TransactionWithStatusMeta;
const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111");
// 根据实际并发量调整通道大小，避免背压
const CHANNEL_SIZE: usize = 50000; // 增加到 50000
#[derive(Debug)]
pub enum SystemEvent {
@@ -51,7 +49,6 @@ impl YellowstoneGrpc {
.subscription_manager
.subscribe_with_request(transactions, None, None, None)
.await?;
let (mut tx, mut rx) = mpsc::channel::<EventPretty>(CHANNEL_SIZE);
let callback = Box::new(callback);
@@ -59,16 +56,17 @@ impl YellowstoneGrpc {
while let Some(message) = stream.next().await {
match message {
Ok(msg) => {
if let Err(e) = StreamHandler::handle_stream_message(
msg,
&mut tx,
&mut subscribe_tx,
BackpressureStrategy::Block,
)
.await
if let Ok(event_pretty) =
StreamHandler::handle_stream_system_message(msg, &mut subscribe_tx)
.await
{
error!("Error handling message: {e:?}");
break;
if let Some(event_pretty) = event_pretty {
if let Err(e) =
Self::process_system_transaction(event_pretty, &*callback).await
{
error!("Error processing transaction: {e:?}");
}
}
}
}
Err(error) => {
@@ -78,12 +76,6 @@ impl YellowstoneGrpc {
}
}
});
while let Some(event_pretty) = rx.next().await {
if let Err(e) = Self::process_system_transaction(event_pretty, &*callback).await {
error!("Error processing transaction: {e:?}");
}
}
Ok(())
}