diff --git a/README.md b/README.md index 88f8099..3353776 100755 --- a/README.md +++ b/README.md @@ -17,9 +17,14 @@ A lightweight Rust library for real-time event streaming from Solana DEX trading 5. **Unified Event Interface**: Consistent event handling across all supported protocols 6. **Event Parsing System**: Automatic parsing and categorization of protocol-specific events 7. **High Performance**: Optimized for low-latency event processing -8. **Batch Processing**: Efficient event batching to improve throughput and reduce overhead -9. **Performance Monitoring**: Built-in performance metrics and monitoring capabilities -10. **Memory Optimization**: Object pooling and caching to reduce memory allocations +8. **Batch Processing Optimization**: Batch processing events to reduce callback overhead +9. **Performance Monitoring**: Built-in performance metrics monitoring, including event processing speed, memory usage, etc. +10. **Memory Optimization**: Object pooling and caching mechanisms to reduce memory allocations +11. **Flexible Configuration System**: Support for custom batch sizes, backpressure strategies, channel sizes, and other parameters +12. **Preset Configurations**: Provides high-performance, low-latency, ordered processing, and other preset configurations +13. **Backpressure Handling**: Supports blocking, dropping, retrying, ordered, and other backpressure strategies +14. **Runtime Configuration Updates**: Supports dynamic configuration parameter updates at runtime +15. **Full Function Performance Monitoring**: All subscribe_events functions support performance monitoring, automatically collecting and reporting performance metrics ## Installation @@ -48,7 +53,7 @@ solana-streamer-sdk = "0.1.9" ## Usage Examples -### Basic Usage with Performance Monitoring +### Advanced Usage with Batch Processing and Backpressure ```rust use solana_streamer_sdk::{ @@ -89,11 +94,14 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // Create gRPC client with performance monitoring enabled + // Create low-latency configuration + let mut config = ClientConfig::low_latency(); + // Enable performance monitoring, has performance overhead, disabled by default + config.enable_metrics = true; let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, - true, // enable performance monitoring + config, )?; let callback = create_event_callback(); @@ -109,11 +117,12 @@ async fn test_grpc() -> Result<(), Box> { // Configure account filtering 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 + PUMPFUN_PROGRAM_ID.to_string(), // Monitor pumpfun program ID + PUMPSWAP_PROGRAM_ID.to_string(), // Monitor pumpswap program ID + BONK_PROGRAM_ID.to_string(), // Monitor bonk program ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Monitor raydium_cpmm program ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Monitor raydium_clmm program ID + "xxxxxxxx".to_string(), // Monitor xxxxx account ]; let account_exclude = vec![]; let account_required = vec![]; @@ -141,11 +150,13 @@ async fn test_grpc() -> Result<(), Box> { async fn test_shreds() -> Result<(), Box> { println!("Subscribing to ShredStream events..."); - // Create ShredStream client with performance monitoring enabled - let shred_stream = ShredStreamGrpc::new_with_config( - "http://127.0.0.1:10800".to_string(), - true, // enable performance monitoring - ).await?; + // Create low-latency configuration + let mut config = ShredClientConfig::low_latency(); + // 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?; + let callback = create_event_callback(); let protocols = vec![ Protocol::PumpFun, @@ -275,136 +286,6 @@ src/ └── main.rs # Example program ``` -## Performance Optimizations - -### Recent Performance Improvements - -The latest version includes significant performance optimizations that dramatically improve event processing throughput: - -#### 1. **Batch Processing System** -- **Event Batching**: Events are now processed in batches (default: 100 events per batch) instead of individually -- **Reduced Callback Overhead**: Batch processing reduces the number of callback invocations by up to 100x -- **Improved Throughput**: Significantly higher event processing rates with lower CPU usage -- **Configurable Batch Size**: Adjustable batch size to balance latency vs throughput - -#### 2. **Memory Optimization** -- **Object Pooling**: `EventMetadataPool` and `TransferDataPool` reduce memory allocations -- **Pre-allocated Vectors**: `Vec::with_capacity()` for collections to avoid dynamic resizing -- **Reduced Cloning**: Minimized unnecessary data cloning operations -- **Memory Usage Monitoring**: Real-time memory usage tracking in performance metrics - -#### 3. **Caching System** -- **Event Parse Cache**: `EventParseCache` avoids redundant transaction parsing -- **Cache Hit Rate Monitoring**: Track cache effectiveness in performance metrics -- **Intelligent Cache Management**: Automatic cache size management - -#### 4. **Performance Monitoring** -- **Real-time Metrics**: Built-in performance monitoring with automatic display -- **Comprehensive Statistics**: Events/second, processing times, memory usage, cache hit rates -- **Configurable Monitoring**: Enable/disable performance monitoring as needed -- **Zero Overhead**: Monitoring can be completely disabled for maximum performance - -#### 5. **Concurrent Processing** -- **Async Event Processing**: Non-blocking event handling with `tokio` -- **Parallel Protocol Parsing**: Multiple protocols parsed concurrently -- **Optimized Channel Sizes**: Increased channel capacity (5000) to handle high event volumes - -### Performance Metrics - -The built-in performance monitoring provides detailed insights: - -- **Events Processed**: Total number of events processed -- **Events/Second**: Real-time processing rate (5-second rolling window) -- **Average Processing Time**: Mean time to process events -- **Min/Max Processing Time**: Fastest and slowest processing times -- **Cache Hit Rate**: Percentage of cache hits for event parsing -- **Memory Usage**: Estimated memory consumption - -### Configuration Options - -```rust -// Performance monitoring configuration -let grpc = YellowstoneGrpc::new_with_config( - endpoint, - x_token, - true, // enable performance monitoring -)?; - -// Batch processing is automatically enabled with optimal settings -// Batch size: 100 events -// Batch timeout: 10ms -// Channel size: 5000 -``` - -## Performance Considerations - -1. **Connection Management**: Properly handle connection lifecycle and reconnection -2. **Event Filtering**: Use protocol filtering to reduce unnecessary event processing -3. **Memory Management**: Implement proper cleanup for long-running streams -4. **Error Handling**: Robust error handling for network issues and service disruptions -5. **Batch Processing**: Leverage batch processing for high-throughput scenarios -6. **Performance Monitoring**: Use built-in metrics to optimize your application - -## Configuration Options - -### Yellowstone gRPC Configuration - -```rust -// Recommended: Create gRPC client with performance monitoring enabled -let grpc = YellowstoneGrpc::new_with_config( - "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), - None, - true, // enable performance monitoring -)?; - -// Alternative: Basic configuration (performance monitoring enabled by default) -let grpc = YellowstoneGrpc::new( - "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), - None, -)?; - -// Maximum performance: Disable performance monitoring -let grpc = YellowstoneGrpc::new_with_config( - "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), - None, - false, // disable performance monitoring -)?; -``` - -### ShredStream Configuration - -```rust -// Recommended: Create ShredStream client with performance monitoring enabled -let shred_stream = ShredStreamGrpc::new_with_config( - "http://127.0.0.1:10800".to_string(), - true, // enable performance monitoring -).await?; - -// Alternative: Basic configuration (performance monitoring enabled by default) -let shred_stream = ShredStreamGrpc::new("http://127.0.0.1:10800".to_string()).await?; - -// Maximum performance: Disable performance monitoring -let shred_stream = ShredStreamGrpc::new_with_config( - "http://127.0.0.1:10800".to_string(), - false, // disable performance monitoring -).await?; -``` - -### Performance Tuning - -```rust -// Runtime performance monitoring control -grpc.set_enable_metrics(true).await; // Enable monitoring -grpc.set_enable_metrics(false).await; // Disable monitoring - -// Get current performance metrics -let metrics = grpc.get_metrics().await; -println!("Current performance: {:?}", metrics); - -// Manual performance metrics display -grpc.print_metrics().await; -``` - ## License MIT License @@ -414,6 +295,15 @@ MIT License - Project Repository: https://github.com/0xfnzero/solana-streamer - Telegram Group: https://t.me/fnzero_group +## Performance Considerations + +1. **Connection Management**: Properly handle connection lifecycle and reconnection +2. **Event Filtering**: Use protocol filtering to reduce unnecessary event processing +3. **Memory Management**: Implement appropriate cleanup for long-running streams +4. **Error Handling**: Robust error handling for network issues and service interruptions +5. **Batch Processing Optimization**: Use batch processing to reduce callback overhead and improve throughput +6. **Performance Monitoring**: Enable performance monitoring to identify bottlenecks and optimization opportunities + ## Important Notes 1. **Network Stability**: Ensure stable network connection for continuous event streaming diff --git a/README_CN.md b/README_CN.md index f2eb324..5cefe3e 100644 --- a/README_CN.md +++ b/README_CN.md @@ -20,6 +20,11 @@ 8. **批处理优化**: 批量处理事件以减少回调开销 9. **性能监控**: 内置性能指标监控，包括事件处理速度、内存使用等 10. **内存优化**: 对象池和缓存机制减少内存分配 +11. **灵活配置系统**: 支持自定义批处理大小、背压策略、通道大小等参数 +12. **预设配置**: 提供高性能、低延迟、有序处理等预设配置 +13. **背压处理**: 支持阻塞、丢弃、重试、有序等多种背压策略 +14. **运行时配置更新**: 支持在运行时动态更新配置参数 +15. **全函数性能监控**: 所有subscribe_events函数都支持性能监控，自动收集和报告性能指标 ## 安装 @@ -81,11 +86,14 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("正在订阅 Yellowstone gRPC 事件..."); - // 创建 gRPC 客户端并启用性能监控 + // 创建低延迟配置 + let mut config = ClientConfig::low_latency(); + // 启用性能监控, 有性能损耗, 默认关闭 + config.enable_metrics = true; let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, - true, // 启用性能监控 + config, )?; let callback = create_event_callback(); @@ -128,12 +136,12 @@ async fn test_grpc() -> Result<(), Box> { async fn test_shreds() -> Result<(), Box> { println!("正在订阅 ShredStream 事件..."); - - // 创建 ShredStream 客户端并启用性能监控 - let shred_stream = ShredStreamGrpc::new_with_config( - "http://127.0.0.1:10800".to_string(), - true, // 启用性能监控 - ).await?; + // 创建低延迟配置 + let mut config = ShredClientConfig::low_latency(); + // 启用性能监控, 有性能损耗, 默认关闭 + config.enable_metrics = true; + let shred_stream = + ShredStreamGrpc::new_with_config("http://127.0.0.1:10800".to_string(), config).await?; let callback = create_event_callback(); let protocols = vec![ Protocol::PumpFun, @@ -269,51 +277,6 @@ src/ 5. **批处理优化**: 使用批处理减少回调开销，提高吞吐量 6. **性能监控**: 启用性能监控以识别瓶颈和优化机会 -## 配置选项 - -### Yellowstone gRPC 配置 - -```rust -// 推荐：创建 gRPC 客户端并启用性能监控 -let grpc = YellowstoneGrpc::new_with_config( - "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), - None, - true, // 启用性能监控 -)?; - -// 替代：基本配置（性能监控默认启用） -let grpc = YellowstoneGrpc::new( - "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), - None, -)?; - -// 最大性能：禁用性能监控 -let grpc = YellowstoneGrpc::new_with_config( - "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), - None, - false, // 禁用性能监控 -)?; -``` - -### ShredStream 配置 - -```rust -// 推荐：创建 ShredStream 客户端并启用性能监控 -let shred_stream = ShredStreamGrpc::new_with_config( - "http://127.0.0.1:10800".to_string(), - true, // 启用性能监控 -).await?; - -// 替代：基本配置（性能监控默认启用） -let shred_stream = ShredStreamGrpc::new("http://127.0.0.1:10800".to_string()).await?; - -// 最大性能：禁用性能监控 -let shred_stream = ShredStreamGrpc::new_with_config( - "http://127.0.0.1:10800".to_string(), - false, // 禁用性能监控 -).await?; -``` - ## 许可证 MIT 许可证 diff --git a/src/main.rs b/src/main.rs index 8403e52..78f3137 100755 --- a/src/main.rs +++ b/src/main.rs @@ -20,6 +20,8 @@ use solana_streamer_sdk::{ Protocol, UnifiedEvent, }, ShredStreamGrpc, YellowstoneGrpc, + yellowstone_grpc::ClientConfig, + shred_stream::ShredClientConfig, }, }; @@ -34,11 +36,14 @@ async fn main() -> Result<(), Box> { async fn test_grpc() -> Result<(), Box> { println!("Subscribing to Yellowstone gRPC events..."); - // enable_metrics 为 true 时，会打印性能指标 + // Create low-latency configuration + let mut config = ClientConfig::low_latency(); + // Enable performance monitoring, has performance overhead, disabled by default + config.enable_metrics = true; let grpc = YellowstoneGrpc::new_with_config( "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), None, - true, + config, )?; println!("GRPC client created successfully"); @@ -89,9 +94,12 @@ async fn test_grpc() -> Result<(), Box> { async fn test_shreds() -> Result<(), Box> { println!("Subscribing to ShredStream events..."); - // enable_metrics 为 true 时，会打印性能指标 + // Create low-latency configuration + let mut config = ShredClientConfig::low_latency(); + // 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(), true).await?; + ShredStreamGrpc::new_with_config("http://127.0.0.1:10800".to_string(), config).await?; let callback = create_event_callback(); let protocols = vec![ @@ -113,7 +121,7 @@ fn create_event_callback() -> impl Fn(Box) { println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); match_event!(event, { BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { - // 使用grpc的时候，可以从每个事件中获取到block_time + // When using grpc, you can get block_time from each event println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); println!("BonkPoolCreateEvent: {:?}", e.base_mint_param.symbol); }, diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 70fed50..3fe05d4 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -15,13 +15,99 @@ use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient use crate::protos::shredstream::SubscribeEntriesRequest; use solana_sdk::pubkey::Pubkey; -// 根据实际并发量调整通道大小，避免背压 -const CHANNEL_SIZE: usize = 5000; +// 默认配置常量 +const DEFAULT_CHANNEL_SIZE: usize = 1000; +const DEFAULT_BATCH_SIZE: usize = 100; +const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5; -// 批处理配置 -const SHRED_BATCH_SIZE: usize = 100; -#[allow(dead_code)] -const SHRED_BATCH_TIMEOUT_MS: u64 = 5; +/// ShredStream批处理配置 +#[derive(Debug, Clone)] +pub struct ShredBatchConfig { + /// 批处理大小（默认：100） + pub batch_size: usize, + /// 批处理超时时间（毫秒，默认：10ms） + pub batch_timeout_ms: u64, + /// 是否启用批处理（默认：true） + pub enabled: bool, +} + +impl Default for ShredBatchConfig { + fn default() -> Self { + Self { + batch_size: DEFAULT_BATCH_SIZE, + batch_timeout_ms: DEFAULT_BATCH_TIMEOUT_MS, + enabled: true, + } + } +} + +/// ShredStream背压配置 +#[derive(Debug, Clone)] +pub struct ShredBackpressureConfig { + /// 通道大小（默认：10000） + pub channel_size: usize, +} + +impl Default for ShredBackpressureConfig { + fn default() -> Self { + Self { + channel_size: DEFAULT_CHANNEL_SIZE, + } + } +} + +/// ShredStream完整配置 +#[derive(Debug, Clone)] +pub struct ShredClientConfig { + /// 批处理配置 + pub batch: ShredBatchConfig, + /// 背压配置 + pub backpressure: ShredBackpressureConfig, + /// 是否启用性能监控（默认：false） + pub enable_metrics: bool, +} + +impl Default for ShredClientConfig { + fn default() -> Self { + Self { + batch: ShredBatchConfig::default(), + backpressure: ShredBackpressureConfig::default(), + enable_metrics: false, + } + } +} + +impl ShredClientConfig { + /// 创建高性能配置（适合高并发场景） + pub fn high_performance() -> Self { + Self { + batch: ShredBatchConfig { + batch_size: 200, + batch_timeout_ms: 5, + enabled: true, + }, + backpressure: ShredBackpressureConfig { + channel_size: 20000, + }, + enable_metrics: true, + } + } + + /// 创建低延迟配置（适合实时场景） + pub fn low_latency() -> Self { + Self { + batch: ShredBatchConfig { + batch_size: 10, + batch_timeout_ms: 1, + enabled: false, // 禁用批处理，即时处理 + }, + backpressure: ShredBackpressureConfig { + channel_size: 1000, + }, + enable_metrics: false, + } + } +} /// ShredStream性能监控指标 #[derive(Debug, Clone)] @@ -63,8 +149,8 @@ impl ShredPerformanceMetrics { #[derive(Clone)] pub struct ShredStreamGrpc { shredstream_client: Arc>, + config: ShredClientConfig, metrics: Arc>, - enable_metrics: bool, // 是否启用性能监控 } struct TransactionWithSlot { @@ -80,24 +166,29 @@ where callback: F, batch: Vec>, batch_size: usize, + timeout_ms: u64, + last_flush_time: std::time::Instant, } impl ShredBatchProcessor where F: FnMut(Vec>) + Send + Sync + 'static, { - pub fn new(callback: F, batch_size: usize) -> Self { + pub fn new(callback: F, batch_size: usize, timeout_ms: u64) -> Self { Self { callback, batch: Vec::with_capacity(batch_size), batch_size, + timeout_ms, + last_flush_time: std::time::Instant::now(), } } pub fn add_event(&mut self, event: Box) { self.batch.push(event); - if self.batch.len() >= self.batch_size { + // 检查是否需要刷新批次 + if self.batch.len() >= self.batch_size || self.should_flush_by_timeout() { self.flush(); } } @@ -106,24 +197,51 @@ where if !self.batch.is_empty() { let events = std::mem::replace(&mut self.batch, Vec::with_capacity(self.batch_size)); (self.callback)(events); + self.last_flush_time = std::time::Instant::now(); } } + + fn should_flush_by_timeout(&self) -> bool { + self.last_flush_time.elapsed().as_millis() >= self.timeout_ms as u128 + } } impl ShredStreamGrpc { + /// 创建客户端，使用默认配置 pub async fn new(endpoint: String) -> AnyResult { - Self::new_with_config(endpoint, true).await + Self::new_with_config(endpoint, ShredClientConfig::default()).await } - pub async fn new_with_config(endpoint: String, enable_metrics: bool) -> AnyResult { + /// 创建客户端，使用自定义配置 + pub async fn new_with_config(endpoint: String, config: ShredClientConfig) -> AnyResult { let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?; Ok(Self { shredstream_client: Arc::new(shredstream_client), + config, metrics: Arc::new(Mutex::new(ShredPerformanceMetrics::new())), - enable_metrics, }) } + /// 创建高性能客户端（适合高并发场景） + pub async fn new_high_performance(endpoint: String) -> AnyResult { + Self::new_with_config(endpoint, ShredClientConfig::high_performance()).await + } + + /// 创建低延迟客户端（适合实时场景） + pub async fn new_low_latency(endpoint: String) -> AnyResult { + Self::new_with_config(endpoint, ShredClientConfig::low_latency()).await + } + + /// 获取当前配置 + pub fn get_config(&self) -> &ShredClientConfig { + &self.config + } + + /// 更新配置 + pub fn update_config(&mut self, config: ShredClientConfig) { + self.config = config; + } + /// 获取性能指标 pub async fn get_metrics(&self) -> ShredPerformanceMetrics { let metrics = self.metrics.lock().await; @@ -132,7 +250,7 @@ impl ShredStreamGrpc { /// 启用或禁用性能监控 pub fn set_enable_metrics(&mut self, enabled: bool) { - self.enable_metrics = enabled; + self.config.enable_metrics = enabled; } /// 打印性能指标 @@ -151,7 +269,7 @@ impl ShredStreamGrpc { /// 启动自动性能监控任务 pub async fn start_auto_metrics_monitoring(&self) { // 检查是否启用性能监控 - if !self.enable_metrics { + if !self.config.enable_metrics { return; // 如果未启用性能监控，不启动监控任务 } @@ -168,7 +286,7 @@ impl ShredStreamGrpc { /// 更新性能指标 async fn update_metrics(&self, events_processed: u64, processing_time_ms: f64) { // 检查是否启用性能监控 - if !self.enable_metrics { + if !self.config.enable_metrics { return; // 如果未启用性能监控，直接返回 } @@ -217,6 +335,7 @@ impl ShredStreamGrpc { metrics.memory_usage_mb = metrics.events_processed as f64 * 0.001; // 每个事件约1KB } + /// 订阅ShredStream事件（支持批处理和即时处理） pub async fn shredstream_subscribe( &self, protocols: Vec, @@ -226,14 +345,39 @@ impl ShredStreamGrpc { where F: Fn(Box) + Send + Sync + 'static, { - // 启动自动性能监控 - self.start_auto_metrics_monitoring().await; + // 启动自动性能监控（如果启用） + if self.config.enable_metrics { + self.start_auto_metrics_monitoring().await; + } let request = tonic::Request::new(SubscribeEntriesRequest {}); let mut client = (*self.shredstream_client).clone(); - let mut stream = client.subscribe_entries(request).await?.into_inner(); - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + let stream = client.subscribe_entries(request).await?.into_inner(); + let (tx, rx) = mpsc::channel::(self.config.backpressure.channel_size); + // 根据配置选择处理模式 + if self.config.batch.enabled { + // 批处理模式 + self.process_with_batch(stream, tx, rx, protocols, bot_wallet, callback).await + } else { + // 即时处理模式 + self.process_immediate(stream, tx, rx, protocols, bot_wallet, callback).await + } + } + + /// 批处理模式 + async fn process_with_batch( + &self, + mut stream: tonic::codec::Streaming, + mut tx: mpsc::Sender, + mut rx: mpsc::Receiver, + protocols: Vec, + bot_wallet: Option, + callback: F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync + 'static, + { // 创建批处理器，将单个事件回调转换为批量回调 let batch_callback = move |events: Vec>| { for event in events { @@ -241,7 +385,11 @@ impl ShredStreamGrpc { } }; - let mut batch_processor = ShredBatchProcessor::new(batch_callback, SHRED_BATCH_SIZE); + let mut batch_processor = ShredBatchProcessor::new( + batch_callback, + self.config.batch.batch_size, + self.config.batch.batch_timeout_ms + ); tokio::spawn(async move { while let Some(message) = stream.next().await { @@ -286,6 +434,119 @@ impl ShredStreamGrpc { Ok(()) } + /// 即时处理模式 + async fn process_immediate( + &self, + mut stream: tonic::codec::Streaming, + mut tx: mpsc::Sender, + mut rx: mpsc::Receiver, + protocols: Vec, + bot_wallet: Option, + callback: F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync + 'static, + { + tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Ok(entries) = bincode::deserialize::>(&msg.entries) { + for entry in entries { + for transaction in entry.transactions { + let _ = tx.try_send(TransactionWithSlot { + transaction: transaction.clone(), + slot: msg.slot, + }); + } + } + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); + + let self_clone = self.clone(); + while let Some(transaction_with_slot) = rx.next().await { + if let Err(e) = self_clone.process_transaction_immediate( + transaction_with_slot, + protocols.clone(), + bot_wallet, + &callback, + ) + .await + { + error!("Error processing transaction: {e:?}"); + } + } + + Ok(()) + } + + /// 即时处理单个交易 + async fn process_transaction_immediate( + &self, + transaction_with_slot: TransactionWithSlot, + protocols: Vec, + bot_wallet: Option, + callback: &F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync, + { + let start_time = std::time::Instant::now(); + let program_received_time_ms = chrono::Utc::now().timestamp_millis(); + let slot = transaction_with_slot.slot; + let versioned_tx = transaction_with_slot.transaction; + let signature = versioned_tx.signatures[0]; + + // 预分配向量容量 + let mut all_events = Vec::with_capacity(protocols.len() * 2); + + for protocol in protocols { + let parser = EventParserFactory::create_parser(protocol.clone()); + let events = parser + .parse_versioned_transaction( + &versioned_tx, + &signature.to_string(), + Some(slot), + None, + program_received_time_ms, + bot_wallet, + ) + .await + .unwrap_or_else(|_e| vec![]); + all_events.extend(events); + } + + // 保存事件数量用于日志记录 + let event_count = all_events.len(); + + // 即时处理事件 + for event in all_events { + callback(event); + } + + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + + // 实际调用性能指标更新 + self.update_metrics(event_count as u64, processing_time_ms).await; + + // 记录慢处理操作 + if processing_time_ms > 5.0 { + log::warn!("ShredStream transaction processing took {}ms for {} events", + processing_time_ms, event_count); + } + + Ok(()) + } + async fn process_transaction_with_batch( &self, transaction_with_slot: TransactionWithSlot, diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 25e9af7..5c008bd 100755 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -22,37 +22,166 @@ use crate::streaming::event_parser::{EventParserFactory, Protocol, UnifiedEvent} type TransactionsFilterMap = HashMap; -const CONNECT_TIMEOUT: u64 = 10; -const REQUEST_TIMEOUT: u64 = 60; -// 根据实际并发量调整通道大小，避免背压 -const CHANNEL_SIZE: usize = 5000; -const MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10; +// 默认配置常量 +const DEFAULT_CONNECT_TIMEOUT: u64 = 10; +const DEFAULT_REQUEST_TIMEOUT: u64 = 60; +const DEFAULT_CHANNEL_SIZE: usize = 1000; +const DEFAULT_MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10; +const DEFAULT_BATCH_SIZE: usize = 100; +const DEFAULT_BATCH_TIMEOUT_MS: u64 = 5; -// 批处理配置 -const BATCH_SIZE: usize = 100; // 批处理50个事件 -const BATCH_TIMEOUT_MS: u64 = 10; // 减少超时时间到10ms - -// 连接池配置（为将来扩展保留） -#[allow(dead_code)] -const CONNECTION_POOL_SIZE: usize = 5; -#[allow(dead_code)] -const CONNECTION_IDLE_TIMEOUT: Duration = Duration::from_secs(300); // 5分钟 - -// 工作线程池配置 -const WORKER_THREADS: usize = 8; -const TASK_QUEUE_SIZE: usize = 10000; - -/// 工作线程池配置 -pub struct WorkerPoolConfig { - pub worker_threads: usize, - pub task_queue_size: usize, +// 背压处理策略 +#[derive(Debug, Clone, Copy)] +pub enum BackpressureStrategy { + /// 阻塞等待（默认） + Block, + /// 丢弃消息 + Drop, + /// 重试有限次数后丢弃 + Retry { max_attempts: usize, wait_ms: u64 }, + /// 有序处理（确保按 slot 顺序处理） + Ordered { max_pending_slots: usize }, } -impl Default for WorkerPoolConfig { +impl Default for BackpressureStrategy { + fn default() -> Self { + Self::Block + } +} + +/// 批处理配置 +#[derive(Debug, Clone)] +pub struct BatchConfig { + /// 批处理大小（默认：100） + pub batch_size: usize, + /// 批处理超时时间（毫秒，默认：10ms） + pub batch_timeout_ms: u64, + /// 是否启用批处理（默认：true） + pub enabled: bool, +} + +impl Default for BatchConfig { fn default() -> Self { Self { - worker_threads: WORKER_THREADS, - task_queue_size: TASK_QUEUE_SIZE, + batch_size: DEFAULT_BATCH_SIZE, + batch_timeout_ms: DEFAULT_BATCH_TIMEOUT_MS, + enabled: true, + } + } +} + +/// 背压配置 +#[derive(Debug, Clone)] +pub struct BackpressureConfig { + /// 通道大小（默认：10000） + pub channel_size: usize, + /// 背压处理策略（默认：Block） + pub strategy: BackpressureStrategy, +} + +impl Default for BackpressureConfig { + fn default() -> Self { + Self { + channel_size: DEFAULT_CHANNEL_SIZE, + strategy: BackpressureStrategy::default(), + } + } +} + +/// 连接配置 +#[derive(Debug, Clone)] +pub struct ConnectionConfig { + /// 连接超时时间（秒，默认：10） + pub connect_timeout: u64, + /// 请求超时时间（秒，默认：60） + pub request_timeout: u64, + /// 最大解码消息大小（字节，默认：10MB） + pub max_decoding_message_size: usize, +} + +impl Default for ConnectionConfig { + fn default() -> Self { + Self { + connect_timeout: DEFAULT_CONNECT_TIMEOUT, + request_timeout: DEFAULT_REQUEST_TIMEOUT, + max_decoding_message_size: DEFAULT_MAX_DECODING_MESSAGE_SIZE, + } + } +} + +/// 完整的客户端配置 +#[derive(Debug, Clone)] +pub struct ClientConfig { + /// 连接配置 + pub connection: ConnectionConfig, + /// 批处理配置 + pub batch: BatchConfig, + /// 背压配置 + pub backpressure: BackpressureConfig, + /// 是否启用性能监控（默认：false） + pub enable_metrics: bool, +} + +impl Default for ClientConfig { + fn default() -> Self { + Self { + connection: ConnectionConfig::default(), + batch: BatchConfig::default(), + backpressure: BackpressureConfig::default(), + enable_metrics: false, + } + } +} + +impl ClientConfig { + /// 创建高性能配置（适合高并发场景） + pub fn high_performance() -> Self { + Self { + connection: ConnectionConfig::default(), + batch: BatchConfig { + batch_size: 200, + batch_timeout_ms: 5, + enabled: true, + }, + backpressure: BackpressureConfig { + channel_size: 20000, + strategy: BackpressureStrategy::Drop, + }, + enable_metrics: true, + } + } + + /// 创建低延迟配置（适合实时场景） + pub fn low_latency() -> Self { + Self { + connection: ConnectionConfig::default(), + batch: BatchConfig { + batch_size: 10, + batch_timeout_ms: 1, + enabled: false, + }, + backpressure: BackpressureConfig { + channel_size: 1000, + strategy: BackpressureStrategy::Block, + }, + enable_metrics: false, + } + } + + /// 创建有序处理配置（确保事件按顺序处理） + pub fn ordered_processing(max_pending_slots: usize) -> Self { + Self { + connection: ConnectionConfig::default(), + batch: BatchConfig { + batch_size: 50, + batch_timeout_ms: 5, + enabled: true, + }, + backpressure: BackpressureConfig { + channel_size: 15000, + strategy: BackpressureStrategy::Ordered { max_pending_slots }, + }, + enable_metrics: true, } } } @@ -114,9 +243,9 @@ impl GrpcConnectionPool { let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? .x_token(self.x_token.clone())? .tls_config(ClientTlsConfig::new().with_native_roots())? - .max_decoding_message_size(MAX_DECODING_MESSAGE_SIZE) - .connect_timeout(Duration::from_secs(CONNECT_TIMEOUT)) - .timeout(Duration::from_secs(REQUEST_TIMEOUT)); + .max_decoding_message_size(DEFAULT_MAX_DECODING_MESSAGE_SIZE) + .connect_timeout(Duration::from_secs(DEFAULT_CONNECT_TIMEOUT)) + .timeout(Duration::from_secs(DEFAULT_REQUEST_TIMEOUT)); Ok(builder.connect().await?) } @@ -127,7 +256,7 @@ pub struct EventBatchCollector where F: Fn(Vec>) + Send + Sync + 'static, { - callback: F, + pub(crate) callback: F, batch: Vec>, batch_size: usize, timeout_ms: u64, @@ -246,20 +375,18 @@ impl From<(SubscribeUpdateTransaction, Option)> for TransactionPretty pub struct YellowstoneGrpc { endpoint: String, x_token: Option, + config: ClientConfig, metrics: Arc>, - enable_metrics: bool, // 是否启用性能监控 } impl YellowstoneGrpc { + /// 创建客户端，使用默认配置 pub fn new(endpoint: String, x_token: Option) -> AnyResult { - Self::new_with_config(endpoint, x_token, true) + Self::new_with_config(endpoint, x_token, ClientConfig::default()) } - pub fn new_with_config( - endpoint: String, - x_token: Option, - enable_metrics: bool, - ) -> AnyResult { + /// 创建客户端，使用自定义配置 + pub fn new_with_config(endpoint: String, x_token: Option, config: ClientConfig) -> AnyResult { if CryptoProvider::get_default().is_none() { default_provider() .install_default() @@ -269,11 +396,43 @@ impl YellowstoneGrpc { Ok(Self { endpoint, x_token, + config, metrics: Arc::new(Mutex::new(PerformanceMetrics::new())), - enable_metrics, }) } + /// 创建高性能客户端（适合高并发场景） + pub fn new_high_performance(endpoint: String, x_token: Option) -> AnyResult { + Self::new_with_config(endpoint, x_token, ClientConfig::high_performance()) + } + + /// 创建低延迟客户端（适合实时场景） + pub fn new_low_latency(endpoint: String, x_token: Option) -> AnyResult { + Self::new_with_config(endpoint, x_token, ClientConfig::low_latency()) + } + + /// 创建有序处理客户端（确保事件按顺序处理） + pub fn new_ordered_processing(endpoint: String, x_token: Option, max_pending_slots: usize) -> AnyResult { + Self::new_with_config(endpoint, x_token, ClientConfig::ordered_processing(max_pending_slots)) + } + + /// 创建简化的即时处理客户端（推荐用于简单场景） + pub fn new_immediate(endpoint: String, x_token: Option) -> AnyResult { + let mut config = ClientConfig::low_latency(); + config.enable_metrics = false; // 即时模式默认关闭性能监控 + Self::new_with_config(endpoint, x_token, config) + } + + /// 获取当前配置 + pub fn get_config(&self) -> &ClientConfig { + &self.config + } + + /// 更新配置 + pub fn update_config(&mut self, config: ClientConfig) { + self.config = config; + } + /// 获取性能指标 pub async fn get_metrics(&self) -> PerformanceMetrics { let metrics = self.metrics.lock().await; @@ -282,7 +441,7 @@ impl YellowstoneGrpc { /// 启用或禁用性能监控 pub fn set_enable_metrics(&mut self, enabled: bool) { - self.enable_metrics = enabled; + self.config.enable_metrics = enabled; } @@ -304,7 +463,7 @@ impl YellowstoneGrpc { /// 启动自动性能监控任务 pub async fn start_auto_metrics_monitoring(&self) { // 检查是否启用性能监控 - if !self.enable_metrics { + if !self.config.enable_metrics { return; // 如果未启用性能监控，不启动监控任务 } @@ -321,7 +480,7 @@ impl YellowstoneGrpc { /// 更新性能指标 async fn update_metrics(&self, events_processed: u64, processing_time_ms: f64) { // 检查是否启用性能监控 - if !self.enable_metrics { + if !self.config.enable_metrics { return; // 如果未启用性能监控，直接返回 } @@ -374,9 +533,9 @@ impl YellowstoneGrpc { let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? .x_token(self.x_token.clone())? .tls_config(ClientTlsConfig::new().with_native_roots())? - .max_decoding_message_size(MAX_DECODING_MESSAGE_SIZE) - .connect_timeout(Duration::from_secs(CONNECT_TIMEOUT)) - .timeout(Duration::from_secs(REQUEST_TIMEOUT)); + .max_decoding_message_size(self.config.connection.max_decoding_message_size) + .connect_timeout(Duration::from_secs(self.config.connection.connect_timeout)) + .timeout(Duration::from_secs(self.config.connection.request_timeout)); Ok(builder.connect().await?) } @@ -431,13 +590,64 @@ impl YellowstoneGrpc { msg: SubscribeUpdate, tx: &mut mpsc::Sender, subscribe_tx: &mut (impl Sink + Unpin), + backpressure_strategy: BackpressureStrategy, ) -> AnyResult<()> { let created_at = msg.created_at; match msg.update_oneof { Some(UpdateOneof::Transaction(sut)) => { let transaction_pretty = TransactionPretty::from((sut, created_at)); log::info!("Received transaction: {} at slot {}", transaction_pretty.signature, transaction_pretty.slot); - tx.try_send(transaction_pretty)?; + + // 根据背压策略处理发送 + match backpressure_strategy { + BackpressureStrategy::Block => { + // 阻塞等待，直到有空间 + if let Err(e) = tx.send(transaction_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(transaction_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(transaction_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)); + } + } + } + } + } + BackpressureStrategy::Ordered { max_pending_slots: _ } => { + // 有序处理策略 - 这里暂时使用阻塞策略，实际的有序处理在接收端实现 + if let Err(e) = tx.send(transaction_pretty).await { + log::error!("Failed to send transaction to channel: {:?}", e); + return Err(anyhow::anyhow!("Channel send failed: {:?}", e)); + } + } + } } Some(UpdateOneof::Ping(_)) => { subscribe_tx @@ -486,9 +696,120 @@ impl YellowstoneGrpc { where F: Fn(Box) + Send + Sync + 'static, { + // 启动自动性能监控（如果启用） + if self.config.enable_metrics { + self.start_auto_metrics_monitoring().await; + } - // 启动自动性能监控 - self.start_auto_metrics_monitoring().await; + // 默认使用即时处理模式 + self.subscribe_events_immediate( + protocols, + bot_wallet, + account_include, + account_exclude, + account_required, + commitment, + callback, + ) + .await + } + + /// 简化的即时事件订阅（推荐用于简单场景） + pub async fn subscribe_events_immediate( + &self, + protocols: Vec, + bot_wallet: Option, + account_include: Vec, + account_exclude: Vec, + account_required: Vec, + commitment: Option, + callback: F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync + 'static, + { + // 启动自动性能监控（如果启用） + if self.config.enable_metrics { + self.start_auto_metrics_monitoring().await; + } + + if account_include.is_empty() && account_exclude.is_empty() && account_required.is_empty() { + return Err(anyhow::anyhow!( + "account_include or account_exclude or account_required cannot be empty" + )); + } + + let transactions = + self.get_subscribe_request_filter(account_include, account_exclude, account_required); + + // 订阅事件 + let (mut subscribe_tx, mut stream) = self + .subscribe_with_request(transactions, commitment) + .await?; + + // 创建通道，使用配置中的通道大小 + let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); + + // 启动流处理任务 + let backpressure_strategy = self.config.backpressure.strategy; + tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Err(e) = + Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, backpressure_strategy).await + { + error!("Error handling message: {e:?}"); + break; + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); + + // 即时处理交易，无批处理 + let self_clone = self.clone(); + tokio::spawn(async move { + while let Some(transaction_pretty) = rx.next().await { + if let Err(e) = self_clone.process_event_transaction_with_metrics( + transaction_pretty, + &callback, + bot_wallet, + protocols.clone(), + ) + .await + { + error!("Error processing transaction: {e:?}"); + } + } + }); + + tokio::signal::ctrl_c().await?; + Ok(()) + } + + /// 高级模式订阅（包含批处理和背压处理） + pub async fn subscribe_events_advanced( + &self, + protocols: Vec, + bot_wallet: Option, + account_include: Vec, + account_exclude: Vec, + account_required: Vec, + commitment: Option, + callback: F, + ) -> AnyResult<()> + where + F: Fn(Box) + Send + Sync + 'static, + { + // 启动自动性能监控（如果启用） + if self.config.enable_metrics { + self.start_auto_metrics_monitoring().await; + } if account_include.is_empty() && account_exclude.is_empty() && account_required.is_empty() { return Err(anyhow::anyhow!( @@ -504,7 +825,7 @@ impl YellowstoneGrpc { .await?; // Create channel - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); // 创建批处理器，将单个事件回调转换为批量回调 let batch_callback = move |events: Vec>| { @@ -513,15 +834,20 @@ impl YellowstoneGrpc { } }; - let mut batch_processor = EventBatchCollector::new(batch_callback, BATCH_SIZE, BATCH_TIMEOUT_MS); + let mut batch_processor = EventBatchCollector::new( + batch_callback, + self.config.batch.batch_size, + self.config.batch.batch_timeout_ms + ); // Start task to process the stream + let backpressure_strategy = self.config.backpressure.strategy; tokio::spawn(async move { while let Some(message) = stream.next().await { match message { Ok(msg) => { if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await + Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, backpressure_strategy).await { error!("Error handling message: {e:?}"); break; @@ -537,23 +863,50 @@ impl YellowstoneGrpc { // Process transactions with batch processing let self_clone = self.clone(); - tokio::spawn(async move { - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = self_clone.process_event_transaction_with_batch( - transaction_pretty, - &mut batch_processor, - bot_wallet, - protocols.clone(), - ) - .await - { - error!("Error processing transaction: {e:?}"); - } + + // 根据背压策略选择处理方式 + match self.config.backpressure.strategy { + BackpressureStrategy::Ordered { max_pending_slots: _ } => { + // 使用有序处理 - 暂时使用普通的批处理方式 + tokio::spawn(async move { + while let Some(transaction_pretty) = rx.next().await { + if let Err(e) = self_clone.process_event_transaction_with_batch( + transaction_pretty, + &mut batch_processor, + bot_wallet, + protocols.clone(), + ) + .await + { + error!("Error processing transaction: {e:?}"); + } + } + + // 处理剩余的事件 + batch_processor.flush(); + }); } - - // 处理剩余的事件 - batch_processor.flush(); - }); + _ => { + // 使用原有的批处理方式 + tokio::spawn(async move { + while let Some(transaction_pretty) = rx.next().await { + if let Err(e) = self_clone.process_event_transaction_with_batch( + transaction_pretty, + &mut batch_processor, + bot_wallet, + protocols.clone(), + ) + .await + { + error!("Error processing transaction: {e:?}"); + } + } + + // 处理剩余的事件 + batch_processor.flush(); + }); + } + } tokio::signal::ctrl_c().await?; Ok(()) @@ -578,6 +931,11 @@ impl YellowstoneGrpc { where F: Fn(Box) + Send + Sync + 'static, { + // 启动自动性能监控（如果启用） + if self.config.enable_metrics { + self.start_auto_metrics_monitoring().await; + } + // 创建过滤器 let protocol_accounts = protocols .iter() @@ -599,18 +957,19 @@ impl YellowstoneGrpc { .await?; // 创建通道 - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); // 创建回调函数，使用 Arc 包装以便在多个任务中共享 let callback = std::sync::Arc::new(Box::new(callback)); // 启动处理流的任务 + let backpressure_strategy = self.config.backpressure.strategy; tokio::spawn(async move { while let Some(message) = stream.next().await { match message { Ok(msg) => { if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await + Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, backpressure_strategy).await { error!("Error handling message: {e:?}"); break; @@ -625,9 +984,10 @@ impl YellowstoneGrpc { }); // 处理交易 + let self_clone = self.clone(); tokio::spawn(async move { while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_event_transaction( + if let Err(e) = self_clone.process_event_transaction_with_metrics( transaction_pretty, &**callback, bot_wallet, @@ -644,7 +1004,8 @@ impl YellowstoneGrpc { Ok(()) } - async fn process_event_transaction( + async fn process_event_transaction_with_metrics( + &self, transaction_pretty: TransactionPretty, callback: &F, bot_wallet: Option, @@ -705,6 +1066,11 @@ impl YellowstoneGrpc { let processing_time = start_time.elapsed(); let processing_time_ms = processing_time.as_millis() as f64; + // 更新性能指标（如果启用） + if self.config.enable_metrics { + self.update_metrics(event_count as u64, processing_time_ms).await; + } + // 记录慢处理操作 if processing_time_ms > 10.0 { log::warn!("Slow event processing: {processing_time_ms}ms for {event_count} events"); @@ -778,7 +1144,14 @@ impl YellowstoneGrpc { total_events += events.len(); log::info!("Adding {} events to batch processor", events.len()); for event in events { - batch_processor.add_event(event); + if self.config.batch.enabled { + batch_processor.add_event(event); + } else { + // 如果批处理被禁用，直接调用回调 + // 这里需要将单个事件包装成Vec来调用批处理回调 + let single_event_batch = vec![event]; + (batch_processor.callback)(single_event_batch); + } } } Err(e) => { @@ -813,3 +1186,90 @@ impl YellowstoneGrpc { } } +// 有序交易处理器，确保交易按顺序处理 +pub struct OrderedTransactionProcessor +where + F: Fn(TransactionPretty) + Send + Sync + 'static, +{ + callback: F, + pending_transactions: std::collections::BTreeMap, // 按 slot 排序 + next_expected_slot: u64, + max_pending_slots: usize, // 最大等待槽位数 +} + +impl OrderedTransactionProcessor +where + F: Fn(TransactionPretty) + Send + Sync + 'static, +{ + pub fn new(callback: F, max_pending_slots: usize) -> Self { + Self { + callback, + pending_transactions: std::collections::BTreeMap::new(), + next_expected_slot: 0, + max_pending_slots, + } + } + + pub fn process_transaction(&mut self, transaction: TransactionPretty) { + let slot = transaction.slot; + + // 如果是第一个交易，设置期望的槽位 + if self.next_expected_slot == 0 { + self.next_expected_slot = slot; + } + + // 如果槽位太旧，直接丢弃 + if slot < self.next_expected_slot.saturating_sub(self.max_pending_slots as u64) { + log::warn!("Dropping old transaction from slot {}", slot); + return; + } + + // 如果槽位太新，先缓存 + if slot > self.next_expected_slot { + self.pending_transactions.insert(slot, transaction); + + // 如果缓存太多，清理旧的 + while self.pending_transactions.len() > self.max_pending_slots { + if let Some((oldest_slot, _)) = self.pending_transactions.iter().next() { + let oldest_slot = *oldest_slot; + self.pending_transactions.remove(&oldest_slot); + log::warn!("Dropping old cached transaction from slot {}", oldest_slot); + } + } + return; + } + + // 处理当前槽位的交易 + if slot == self.next_expected_slot { + (self.callback)(transaction); + self.next_expected_slot += 1; + + // 处理后续连续的槽位 + loop { + let next_slot = self.pending_transactions.keys().next().copied(); + if let Some(slot) = next_slot { + if slot == self.next_expected_slot { + if let Some(transaction) = self.pending_transactions.remove(&slot) { + (self.callback)(transaction); + self.next_expected_slot += 1; + } + } else { + break; + } + } else { + break; + } + } + } else { + // 槽位不匹配，缓存起来 + self.pending_transactions.insert(slot, transaction); + } + } + + pub fn get_stats(&self) -> (u64, usize) { + (self.next_expected_slot, self.pending_transactions.len()) + } +} + + + diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index e852780..2062ce5 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -1,6 +1,6 @@ use crate::{ common::AnyResult, - streaming::yellowstone_grpc::{TransactionPretty, YellowstoneGrpc}, + streaming::yellowstone_grpc::{TransactionPretty, YellowstoneGrpc, BackpressureStrategy}, }; use futures::{channel::mpsc, StreamExt}; use log::error; @@ -10,7 +10,7 @@ use solana_transaction_status::EncodedTransactionWithStatusMeta; const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111"); // 根据实际并发量调整通道大小，避免背压 -const CHANNEL_SIZE: usize = 5000; +const CHANNEL_SIZE: usize = 50000; // 增加到 50000 #[derive(Debug)] pub enum SystemEvent { @@ -51,7 +51,7 @@ impl YellowstoneGrpc { match message { Ok(msg) => { if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await + Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx, BackpressureStrategy::Block).await { error!("Error handling message: {e:?}"); break;