From 74781e5cbe6f48877ea564e4ae9da6ec51282b16 Mon Sep 17 00:00:00 2001 From: ysq Date: Sun, 31 Aug 2025 22:18:18 +0800 Subject: [PATCH] perf: implement SIMD-accelerated event processing and optimize streaming performance - Add SIMD utilities for fast byte array comparison and discriminator matching - Optimize event processor with batch processing and memory pool - Refactor global state management with concurrent data structures - Remove deprecated batch processing module - Enhance metrics collection with reduced overhead - Improve parser efficiency across all protocol implementations - Add performance benchmarking dependencies (criterion, wide) - Update documentation and examples for new architecture Performance improvements: - SIMD-accelerated byte operations for instruction parsing - Concurrent HashMap (DashMap) for better multi-threading - Optimized memory allocation patterns - Reduced lock contention in event processing pipeline Breaking changes: Removed batch.rs module, updated parser interfaces --- .gitignore | 3 +- Cargo.toml | 5 + README.md | 32 +- README_CN.md | 34 +- examples/parse_tx_events.rs | 9 +- src/main.rs | 294 +++++++------- src/streaming/common/batch.rs | 131 ------ src/streaming/common/config.rs | 27 +- src/streaming/common/event_processor.rs | 373 ++++++++++-------- src/streaming/common/metrics.rs | 259 +++--------- src/streaming/common/mod.rs | 6 +- src/streaming/common/simd_utils.rs | 295 ++++++++++++++ src/streaming/event_parser/common/mod.rs | 14 +- src/streaming/event_parser/common/types.rs | 46 +-- .../event_parser/core/account_event_parser.rs | 36 +- .../event_parser/core/common_event_parser.rs | 4 +- .../event_parser/core/global_state.rs | 238 +++++++---- src/streaming/event_parser/core/traits.rs | 337 +++++++++++----- .../protocols/block/block_meta_event.rs | 5 +- .../event_parser/protocols/bonk/parser.rs | 28 -- .../event_parser/protocols/pumpfun/parser.rs | 22 -- .../event_parser/protocols/pumpswap/parser.rs | 55 --- .../protocols/raydium_amm_v4/parser.rs | 36 -- .../protocols/raydium_clmm/parser.rs | 24 -- .../protocols/raydium_cpmm/parser.rs | 18 - src/streaming/grpc/subscription.rs | 26 +- src/streaming/shred/connection.rs | 8 - src/streaming/shred_stream.rs | 23 +- src/streaming/yellowstone_grpc.rs | 149 ++++--- src/streaming/yellowstone_sub_system.rs | 19 +- 30 files changed, 1297 insertions(+), 1259 deletions(-) delete mode 100644 src/streaming/common/batch.rs create mode 100644 src/streaming/common/simd_utils.rs diff --git a/.gitignore b/.gitignore index c77fa9e..4433903 100755 --- a/.gitignore +++ b/.gitignore @@ -20,4 +20,5 @@ Cargo.lock # option (not recommended) you can uncomment the following to ignore the entire idea folder. #.idea/ -.cargo/ \ No newline at end of file +.cargo/ +.claude/ \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index d3ed2ce..ce72227 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,6 +51,7 @@ thiserror = "2.0.11" async-trait = "0.1.86" lazy_static = "1.5.0" once_cell = "1.20.3" +dashmap = "6.0.1" prost = "0.13.5" prost-types = "0.13.5" num_enum = "0.7.3" @@ -66,3 +67,7 @@ env_logger = "0.11.8" crossbeam = "0.8.4" crossbeam-queue = "0.3.12" parking_lot = "0.12.1" +wide = "0.7" + +[dev-dependencies] +criterion = { version = "0.5", features = ["html_reports"] } \ No newline at end of file diff --git a/README.md b/README.md index a09432e..011c2a2 100755 --- a/README.md +++ b/README.md @@ -24,8 +24,8 @@ A lightweight Rust library for real-time event streaming from Solana DEX trading 11. **Performance Monitoring**: Built-in performance metrics monitoring, including event processing speed, etc. 12. **Memory Optimization**: Object pooling and caching mechanisms to reduce memory allocations 13. **Flexible Configuration System**: Support for custom batch sizes, backpressure strategies, channel sizes, and other parameters -14. **Preset Configurations**: Provides high-throughput, low-latency, and async processing preset configurations optimized for different use cases -15. **Backpressure Handling**: Supports blocking, dropping, retrying, ordered, and other backpressure strategies +14. **Preset Configurations**: Provides high-throughput and low-latency preset configurations optimized for different use cases +15. **Backpressure Handling**: Supports blocking and dropping backpressure strategies 16. **Runtime Configuration Updates**: Supports dynamic configuration parameter updates at runtime 17. **Full Function Performance Monitoring**: All subscribe_events functions support performance monitoring, automatically collecting and reporting performance metrics 18. **Graceful Shutdown**: Support for programmatic stop() method for clean shutdown @@ -91,26 +91,10 @@ let shred = ShredStreamGrpc::new_low_latency(endpoint).await?; **Features:** - **Backpressure Strategy**: Block - ensures no data loss -- **Buffer Size**: 1 permit to minimize memory usage +- **Buffer Size**: 4000 permits for balanced throughput and latency - **Immediate Processing**: No buffering, processes events immediately - **Use Case**: Scenarios where every millisecond counts and you cannot afford to lose any events, such as trading applications or real-time monitoring -#### 3. Async Processing Configuration (`async_processing()`) - -Balances throughput and reliability: - -```rust -let config = StreamClientConfig::async_processing(); -// Or use convenience methods -let grpc = YellowstoneGrpc::new_async_processing(endpoint, token)?; -let shred = ShredStreamGrpc::new_async_processing(endpoint).await?; -``` - -**Features:** -- **Backpressure Strategy**: Async - non-blocking operation -- **Buffer Size**: 5,000 permits for steady flow -- **Fire-and-forget**: Async processing semantics -- **Use Case**: Scenarios where you need sustained high throughput with eventual consistency, such as data ingestion pipelines or event streaming applications ### Custom Configuration @@ -131,16 +115,6 @@ let config = StreamClientConfig { }; ``` -### Configuration Selection Guide - -| Scenario | Recommended Config | Reason | -|----------|-------------------|---------| -| Trading Bots | `low_latency()` | Need fastest response time, cannot lose trading signals | -| Data Analytics | `high_throughput()` | Need to process large amounts of historical data, can tolerate some data loss | -| Event Stream Processing | `async_processing()` | Balance performance and reliability, suitable for continuous processing | -| Real-time Monitoring | `low_latency()` | Need immediate response to anomalies | -| Bulk Data Ingestion | `high_throughput()` | Prioritize overall throughput | - ## Usage Examples ### Quick Start - Parse Transaction Events diff --git a/README_CN.md b/README_CN.md index cf4330a..e0c94dc 100644 --- a/README_CN.md +++ b/README_CN.md @@ -24,8 +24,8 @@ 11. **性能监控**: 内置性能指标监控，包括事件处理速度等 12. **内存优化**: 对象池和缓存机制减少内存分配 13. **灵活配置系统**: 支持自定义批处理大小、背压策略、通道大小等参数 -14. **预设配置**: 提供高吞吐量、低延迟、异步处理等预设配置，针对不同使用场景优化 -15. **背压处理**: 支持阻塞、丢弃、重试、有序等多种背压策略 +14. **预设配置**: 提供高吞吐量、低延迟等预设配置，针对不同使用场景优化 +15. **背压处理**: 支持阻塞、丢弃等背压策略 16. **运行时配置更新**: 支持在运行时动态更新配置参数 17. **全函数性能监控**: 所有subscribe_events函数都支持性能监控，自动收集和报告性能指标 18. **优雅关闭**: 支持编程式 stop() 方法进行干净的关闭 @@ -90,26 +90,10 @@ let shred = ShredStreamGrpc::new_low_latency(endpoint).await?; **特性：** - **背压策略**: Block（阻塞策略）- 确保不丢失任何数据 -- **缓冲区大小**: 1 个许可证，最小化内存使用 +- **缓冲区大小**: 4000 个许可证，平衡吞吐量和延迟 - **立即处理**: 不进行缓冲，立即处理事件 - **适用场景**: 每毫秒都很重要且不能丢失任何事件的场景，如交易应用或实时监控 -#### 3. 异步处理配置 (`async_processing()`) - -在吞吐量和可靠性之间取得平衡： - -```rust -let config = StreamClientConfig::async_processing(); -// 或者使用便捷方法 -let grpc = YellowstoneGrpc::new_async_processing(endpoint, token)?; -let shred = ShredStreamGrpc::new_async_processing(endpoint).await?; -``` - -**特性：** -- **背压策略**: Async（异步策略）- 非阻塞操作 -- **缓冲区大小**: 5,000 个许可证，保持稳定流量 -- **Fire-and-forget**: 异步处理语义 -- **适用场景**: 需要持续高吞吐量且可接受最终一致性的场景，如数据摄取管道或事件流应用 ### 自定义配置 @@ -130,16 +114,6 @@ let config = StreamClientConfig { }; ``` -### 配置选择指南 - -| 场景 | 推荐配置 | 原因 | -|------|----------|------| -| 交易机器人 | `low_latency()` | 需要最快响应时间，不能丢失交易信号 | -| 数据分析 | `high_throughput()` | 需要处理大量历史数据，可容忍部分数据丢失 | -| 事件流处理 | `async_processing()` | 平衡性能和可靠性，适合持续处理 | -| 实时监控 | `low_latency()` | 需要立即响应异常情况 | -| 批量数据摄取 | `high_throughput()` | 优先考虑整体吞吐量 | - ## 使用示例 ### 快速开始 - 解析交易事件 @@ -587,7 +561,7 @@ let event_type_filter = Some(EventTypeFilter { - **Yellowstone gRPC 客户端**: 针对 Solana 事件流优化 - **ShredStream 客户端**: 替代流实现 -- **异步处理**: 非阻塞事件处理 +- **高性能处理**: 优化的事件处理机制 ## 项目结构 diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index 40b9a02..5f54b8d 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -5,10 +5,6 @@ use solana_streamer_sdk::streaming::event_parser::UnifiedEvent; use solana_streamer_sdk::streaming::event_parser::{ protocols::MutilEventParser, EventParser, Protocol, }; -use solana_transaction_status::{InnerInstruction, InnerInstructions, UiInstruction}; - -use solana_sdk::bs58; -use solana_sdk::instruction::CompiledInstruction; use std::str::FromStr; use std::sync::Arc; @@ -16,7 +12,7 @@ use std::sync::Arc; #[tokio::main] async fn main() -> Result<()> { let signatures = vec![ - "4PsHYajH87x2zJPEGZczZtd2ksibuMCFPonC24jk5mTGZ46hzvjpzM5UZuLz9sRv79MkCBbtDqwJapGPTSkCFKoL", + "5sDWrTTkE69CNc6nrAX7SqPS7FiajJTg8TMog3Gve7KjVfrqYn8YZcX1kAoyKok976S4RTnK1EdCV8hRiDWg68Aj", ]; // Validate signature format let mut valid_signatures = Vec::new(); @@ -119,5 +115,8 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { } } + println!("Press Ctrl+C to exit example..."); + tokio::signal::ctrl_c().await?; + Ok(()) } diff --git a/src/main.rs b/src/main.rs index 51fc068..d6ad5c3 100755 --- a/src/main.rs +++ b/src/main.rs @@ -53,7 +53,7 @@ use solana_streamer_sdk::{ async fn main() -> Result<(), Box> { println!("Starting Solana Streamer..."); test_grpc().await?; - // test_shreds().await?; + test_shreds().await?; Ok(()) } @@ -188,151 +188,151 @@ async fn test_shreds() -> Result<(), Box> { fn create_event_callback() -> impl Fn(Box) { |event: Box| { - // println!( - // "🎉 Event received! Type: {:?}, transaction_index: {:?}", - // event.event_type(), - // event.transaction_index() - // ); - // match_event!(event, { - // // -------------------------- block meta ----------------------- - // BlockMetaEvent => |e: BlockMetaEvent| { - // println!("BlockMetaEvent: {:?}", e.metadata.program_handle_time_consuming_us); - // }, - // // -------------------------- bonk ----------------------- - // BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { - // // 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); - // }, - // BonkTradeEvent => |e: BonkTradeEvent| { - // println!("BonkTradeEvent: {e:?}"); - // }, - // BonkMigrateToAmmEvent => |e: BonkMigrateToAmmEvent| { - // println!("BonkMigrateToAmmEvent: {e:?}"); - // }, - // BonkMigrateToCpswapEvent => |e: BonkMigrateToCpswapEvent| { - // println!("BonkMigrateToCpswapEvent: {e:?}"); - // }, - // // -------------------------- pumpfun ----------------------- - // PumpFunTradeEvent => |e: PumpFunTradeEvent| { - // println!("PumpFunTradeEvent: {e:?}"); - // }, - // PumpFunMigrateEvent => |e: PumpFunMigrateEvent| { - // println!("PumpFunMigrateEvent: {e:?}"); - // }, - // PumpFunCreateTokenEvent => |e: PumpFunCreateTokenEvent| { - // println!("PumpFunCreateTokenEvent: {e:?}"); - // }, - // // -------------------------- pumpswap ----------------------- - // PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { - // println!("Buy event: {e:?}"); - // }, - // PumpSwapSellEvent => |e: PumpSwapSellEvent| { - // println!("Sell event: {e:?}"); - // }, - // PumpSwapCreatePoolEvent => |e: PumpSwapCreatePoolEvent| { - // println!("CreatePool event: {e:?}"); - // }, - // PumpSwapDepositEvent => |e: PumpSwapDepositEvent| { - // println!("Deposit event: {e:?}"); - // }, - // PumpSwapWithdrawEvent => |e: PumpSwapWithdrawEvent| { - // println!("Withdraw event: {e:?}"); - // }, - // // -------------------------- raydium_cpmm ----------------------- - // RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { - // println!("RaydiumCpmmSwapEvent: {e:?}"); - // }, - // RaydiumCpmmDepositEvent => |e: RaydiumCpmmDepositEvent| { - // println!("RaydiumCpmmDepositEvent: {e:?}"); - // }, - // RaydiumCpmmInitializeEvent => |e: RaydiumCpmmInitializeEvent| { - // println!("RaydiumCpmmInitializeEvent: {e:?}"); - // }, - // RaydiumCpmmWithdrawEvent => |e: RaydiumCpmmWithdrawEvent| { - // println!("RaydiumCpmmWithdrawEvent: {e:?}"); - // }, - // // -------------------------- raydium_clmm ----------------------- - // RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { - // println!("RaydiumClmmSwapEvent: {e:?}"); - // }, - // RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { - // println!("RaydiumClmmSwapV2Event: {e:?}"); - // }, - // RaydiumClmmClosePositionEvent => |e: RaydiumClmmClosePositionEvent| { - // println!("RaydiumClmmClosePositionEvent: {e:?}"); - // }, - // RaydiumClmmDecreaseLiquidityV2Event => |e: RaydiumClmmDecreaseLiquidityV2Event| { - // println!("RaydiumClmmDecreaseLiquidityV2Event: {e:?}"); - // }, - // RaydiumClmmCreatePoolEvent => |e: RaydiumClmmCreatePoolEvent| { - // println!("RaydiumClmmCreatePoolEvent: {e:?}"); - // }, - // RaydiumClmmIncreaseLiquidityV2Event => |e: RaydiumClmmIncreaseLiquidityV2Event| { - // println!("RaydiumClmmIncreaseLiquidityV2Event: {e:?}"); - // }, - // RaydiumClmmOpenPositionWithToken22NftEvent => |e: RaydiumClmmOpenPositionWithToken22NftEvent| { - // println!("RaydiumClmmOpenPositionWithToken22NftEvent: {e:?}"); - // }, - // RaydiumClmmOpenPositionV2Event => |e: RaydiumClmmOpenPositionV2Event| { - // println!("RaydiumClmmOpenPositionV2Event: {e:?}"); - // }, - // // -------------------------- raydium_amm_v4 ----------------------- - // RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { - // println!("RaydiumAmmV4SwapEvent: {e:?}"); - // }, - // RaydiumAmmV4DepositEvent => |e: RaydiumAmmV4DepositEvent| { - // println!("RaydiumAmmV4DepositEvent: {e:?}"); - // }, - // RaydiumAmmV4Initialize2Event => |e: RaydiumAmmV4Initialize2Event| { - // println!("RaydiumAmmV4Initialize2Event: {e:?}"); - // }, - // RaydiumAmmV4WithdrawEvent => |e: RaydiumAmmV4WithdrawEvent| { - // println!("RaydiumAmmV4WithdrawEvent: {e:?}"); - // }, - // RaydiumAmmV4WithdrawPnlEvent => |e: RaydiumAmmV4WithdrawPnlEvent| { - // println!("RaydiumAmmV4WithdrawPnlEvent: {e:?}"); - // }, - // // -------------------------- account ----------------------- - // BonkPoolStateAccountEvent => |e: BonkPoolStateAccountEvent| { - // println!("BonkPoolStateAccountEvent: {e:?}"); - // }, - // BonkGlobalConfigAccountEvent => |e: BonkGlobalConfigAccountEvent| { - // println!("BonkGlobalConfigAccountEvent: {e:?}"); - // }, - // BonkPlatformConfigAccountEvent => |e: BonkPlatformConfigAccountEvent| { - // println!("BonkPlatformConfigAccountEvent: {e:?}"); - // }, - // PumpSwapGlobalConfigAccountEvent => |e: PumpSwapGlobalConfigAccountEvent| { - // println!("PumpSwapGlobalConfigAccountEvent: {e:?}"); - // }, - // PumpSwapPoolAccountEvent => |e: PumpSwapPoolAccountEvent| { - // println!("PumpSwapPoolAccountEvent: {e:?}"); - // }, - // PumpFunBondingCurveAccountEvent => |e: PumpFunBondingCurveAccountEvent| { - // println!("PumpFunBondingCurveAccountEvent: {e:?}"); - // }, - // PumpFunGlobalAccountEvent => |e: PumpFunGlobalAccountEvent| { - // println!("PumpFunGlobalAccountEvent: {e:?}"); - // }, - // RaydiumAmmV4AmmInfoAccountEvent => |e: RaydiumAmmV4AmmInfoAccountEvent| { - // println!("RaydiumAmmV4AmmInfoAccountEvent: {e:?}"); - // }, - // RaydiumClmmAmmConfigAccountEvent => |e: RaydiumClmmAmmConfigAccountEvent| { - // println!("RaydiumClmmAmmConfigAccountEvent: {e:?}"); - // }, - // RaydiumClmmPoolStateAccountEvent => |e: RaydiumClmmPoolStateAccountEvent| { - // println!("RaydiumClmmPoolStateAccountEvent: {e:?}"); - // }, - // RaydiumClmmTickArrayStateAccountEvent => |e: RaydiumClmmTickArrayStateAccountEvent| { - // println!("RaydiumClmmTickArrayStateAccountEvent: {e:?}"); - // }, - // RaydiumCpmmAmmConfigAccountEvent => |e: RaydiumCpmmAmmConfigAccountEvent| { - // println!("RaydiumCpmmAmmConfigAccountEvent: {e:?}"); - // }, - // RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| { - // println!("RaydiumCpmmPoolStateAccountEvent: {e:?}"); - // }, - // }); + println!( + "🎉 Event received! Type: {:?}, transaction_index: {:?}", + event.event_type(), + event.transaction_index() + ); + match_event!(event, { + // -------------------------- block meta ----------------------- + BlockMetaEvent => |e: BlockMetaEvent| { + println!("BlockMetaEvent: {:?}", e.metadata.program_handle_time_consuming_us); + }, + // -------------------------- bonk ----------------------- + BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { + // 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); + }, + BonkTradeEvent => |e: BonkTradeEvent| { + println!("BonkTradeEvent: {e:?}"); + }, + BonkMigrateToAmmEvent => |e: BonkMigrateToAmmEvent| { + println!("BonkMigrateToAmmEvent: {e:?}"); + }, + BonkMigrateToCpswapEvent => |e: BonkMigrateToCpswapEvent| { + println!("BonkMigrateToCpswapEvent: {e:?}"); + }, + // -------------------------- pumpfun ----------------------- + PumpFunTradeEvent => |e: PumpFunTradeEvent| { + println!("PumpFunTradeEvent: {e:?}"); + }, + PumpFunMigrateEvent => |e: PumpFunMigrateEvent| { + println!("PumpFunMigrateEvent: {e:?}"); + }, + PumpFunCreateTokenEvent => |e: PumpFunCreateTokenEvent| { + println!("PumpFunCreateTokenEvent: {e:?}"); + }, + // -------------------------- pumpswap ----------------------- + PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { + println!("Buy event: {e:?}"); + }, + PumpSwapSellEvent => |e: PumpSwapSellEvent| { + println!("Sell event: {e:?}"); + }, + PumpSwapCreatePoolEvent => |e: PumpSwapCreatePoolEvent| { + println!("CreatePool event: {e:?}"); + }, + PumpSwapDepositEvent => |e: PumpSwapDepositEvent| { + println!("Deposit event: {e:?}"); + }, + PumpSwapWithdrawEvent => |e: PumpSwapWithdrawEvent| { + println!("Withdraw event: {e:?}"); + }, + // -------------------------- raydium_cpmm ----------------------- + RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { + println!("RaydiumCpmmSwapEvent: {e:?}"); + }, + RaydiumCpmmDepositEvent => |e: RaydiumCpmmDepositEvent| { + println!("RaydiumCpmmDepositEvent: {e:?}"); + }, + RaydiumCpmmInitializeEvent => |e: RaydiumCpmmInitializeEvent| { + println!("RaydiumCpmmInitializeEvent: {e:?}"); + }, + RaydiumCpmmWithdrawEvent => |e: RaydiumCpmmWithdrawEvent| { + println!("RaydiumCpmmWithdrawEvent: {e:?}"); + }, + // -------------------------- raydium_clmm ----------------------- + RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { + println!("RaydiumClmmSwapEvent: {e:?}"); + }, + RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| { + println!("RaydiumClmmSwapV2Event: {e:?}"); + }, + RaydiumClmmClosePositionEvent => |e: RaydiumClmmClosePositionEvent| { + println!("RaydiumClmmClosePositionEvent: {e:?}"); + }, + RaydiumClmmDecreaseLiquidityV2Event => |e: RaydiumClmmDecreaseLiquidityV2Event| { + println!("RaydiumClmmDecreaseLiquidityV2Event: {e:?}"); + }, + RaydiumClmmCreatePoolEvent => |e: RaydiumClmmCreatePoolEvent| { + println!("RaydiumClmmCreatePoolEvent: {e:?}"); + }, + RaydiumClmmIncreaseLiquidityV2Event => |e: RaydiumClmmIncreaseLiquidityV2Event| { + println!("RaydiumClmmIncreaseLiquidityV2Event: {e:?}"); + }, + RaydiumClmmOpenPositionWithToken22NftEvent => |e: RaydiumClmmOpenPositionWithToken22NftEvent| { + println!("RaydiumClmmOpenPositionWithToken22NftEvent: {e:?}"); + }, + RaydiumClmmOpenPositionV2Event => |e: RaydiumClmmOpenPositionV2Event| { + println!("RaydiumClmmOpenPositionV2Event: {e:?}"); + }, + // -------------------------- raydium_amm_v4 ----------------------- + RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { + println!("RaydiumAmmV4SwapEvent: {e:?}"); + }, + RaydiumAmmV4DepositEvent => |e: RaydiumAmmV4DepositEvent| { + println!("RaydiumAmmV4DepositEvent: {e:?}"); + }, + RaydiumAmmV4Initialize2Event => |e: RaydiumAmmV4Initialize2Event| { + println!("RaydiumAmmV4Initialize2Event: {e:?}"); + }, + RaydiumAmmV4WithdrawEvent => |e: RaydiumAmmV4WithdrawEvent| { + println!("RaydiumAmmV4WithdrawEvent: {e:?}"); + }, + RaydiumAmmV4WithdrawPnlEvent => |e: RaydiumAmmV4WithdrawPnlEvent| { + println!("RaydiumAmmV4WithdrawPnlEvent: {e:?}"); + }, + // -------------------------- account ----------------------- + BonkPoolStateAccountEvent => |e: BonkPoolStateAccountEvent| { + println!("BonkPoolStateAccountEvent: {e:?}"); + }, + BonkGlobalConfigAccountEvent => |e: BonkGlobalConfigAccountEvent| { + println!("BonkGlobalConfigAccountEvent: {e:?}"); + }, + BonkPlatformConfigAccountEvent => |e: BonkPlatformConfigAccountEvent| { + println!("BonkPlatformConfigAccountEvent: {e:?}"); + }, + PumpSwapGlobalConfigAccountEvent => |e: PumpSwapGlobalConfigAccountEvent| { + println!("PumpSwapGlobalConfigAccountEvent: {e:?}"); + }, + PumpSwapPoolAccountEvent => |e: PumpSwapPoolAccountEvent| { + println!("PumpSwapPoolAccountEvent: {e:?}"); + }, + PumpFunBondingCurveAccountEvent => |e: PumpFunBondingCurveAccountEvent| { + println!("PumpFunBondingCurveAccountEvent: {e:?}"); + }, + PumpFunGlobalAccountEvent => |e: PumpFunGlobalAccountEvent| { + println!("PumpFunGlobalAccountEvent: {e:?}"); + }, + RaydiumAmmV4AmmInfoAccountEvent => |e: RaydiumAmmV4AmmInfoAccountEvent| { + println!("RaydiumAmmV4AmmInfoAccountEvent: {e:?}"); + }, + RaydiumClmmAmmConfigAccountEvent => |e: RaydiumClmmAmmConfigAccountEvent| { + println!("RaydiumClmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumClmmPoolStateAccountEvent => |e: RaydiumClmmPoolStateAccountEvent| { + println!("RaydiumClmmPoolStateAccountEvent: {e:?}"); + }, + RaydiumClmmTickArrayStateAccountEvent => |e: RaydiumClmmTickArrayStateAccountEvent| { + println!("RaydiumClmmTickArrayStateAccountEvent: {e:?}"); + }, + RaydiumCpmmAmmConfigAccountEvent => |e: RaydiumCpmmAmmConfigAccountEvent| { + println!("RaydiumCpmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| { + println!("RaydiumCpmmPoolStateAccountEvent: {e:?}"); + }, + }); } } diff --git a/src/streaming/common/batch.rs b/src/streaming/common/batch.rs deleted file mode 100644 index f9c092c..0000000 --- a/src/streaming/common/batch.rs +++ /dev/null @@ -1,131 +0,0 @@ -use crate::streaming::event_parser::UnifiedEvent; - -/// 通用批处理事件收集器 -pub struct EventBatchProcessor -where - F: FnMut(Vec>) + Send + Sync + 'static, -{ - pub(crate) callback: F, - batch: Vec>, - batch_size: usize, - timeout_ms: u64, - last_flush_time: std::time::Instant, -} - -impl EventBatchProcessor -where - F: FnMut(Vec>) + Send + Sync + 'static, -{ - /// 创建新的批处理器 - 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) { - log::debug!("Adding event to batch: {} (type: {:?})", event.id(), event.event_type()); - self.batch.push(event); - - // 检查是否需要刷新批次 - if self.batch.len() >= self.batch_size || self.should_flush_by_timeout() { - log::debug!("Flushing batch: size={}, timeout={}", self.batch.len(), self.should_flush_by_timeout()); - self.flush(); - } - } - - /// 强制刷新当前批次 - pub fn flush(&mut self) { - if !self.batch.is_empty() { - let events = std::mem::replace(&mut self.batch, Vec::with_capacity(self.batch_size)); - log::debug!("Flushing {} events from batch processor", events.len()); - - // 添加调试信息（仅在debug模式下） - if log::log_enabled!(log::Level::Debug) { - for (i, event) in events.iter().enumerate() { - log::debug!("Event {}: Type={:?}, ID={}", i, event.event_type(), event.id()); - } - } - - // 执行回调并捕获可能的错误 - log::debug!("Executing batch callback with {} events", events.len()); - match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - (self.callback)(events); - })) { - Ok(_) => { - log::debug!("Batch callback executed successfully"); - } - Err(e) => { - log::error!("Batch callback panicked: {:?}", e); - } - } - - self.last_flush_time = std::time::Instant::now(); - } else { - log::debug!("No events to flush"); - } - } - - /// 获取当前批次大小 - pub fn current_batch_size(&self) -> usize { - self.batch.len() - } - - /// 检查是否应该基于超时刷新 - fn should_flush_by_timeout(&self) -> bool { - self.last_flush_time.elapsed().as_millis() >= self.timeout_ms as u128 - } - - /// 检查批次是否已满 - pub fn is_batch_full(&self) -> bool { - self.batch.len() >= self.batch_size - } - - /// 检查是否需要刷新（大小或超时） - pub fn should_flush(&self) -> bool { - self.is_batch_full() || self.should_flush_by_timeout() - } -} - -/// 简单的事件批处理器，用于将单个事件回调转换为批量回调 -pub struct SimpleEventBatchProcessor -where - F: Fn(Box) + Send + Sync + 'static, -{ - callback: F, -} - -impl SimpleEventBatchProcessor -where - F: Fn(Box) + Send + Sync + 'static, -{ - pub fn new(callback: F) -> Self { - Self { callback } - } - - /// 将批量事件拆分为单个事件处理 - pub fn process_batch(&self, events: Vec>) { - for event in events { - (self.callback)(event); - } - } -} - -/// 批处理器包装器，用于将单个事件回调适配为批量处理 -pub fn create_batch_callback_adapter( - single_event_callback: F, -) -> impl FnMut(Vec>) + Send + Sync + 'static -where - F: Fn(Box) + Send + Sync + 'static, -{ - move |events: Vec>| { - for event in events { - single_event_callback(event); - } - } -} diff --git a/src/streaming/common/config.rs b/src/streaming/common/config.rs index 0c76687..ce21ec8 100644 --- a/src/streaming/common/config.rs +++ b/src/streaming/common/config.rs @@ -7,8 +7,6 @@ pub enum BackpressureStrategy { Block, /// Drop messages Drop, - /// Execute asynchronously (don't wait for completion) - Async, } impl Default for BackpressureStrategy { @@ -87,7 +85,7 @@ impl StreamClientConfig { Self { connection: ConnectionConfig::default(), backpressure: BackpressureConfig { - permits: 5000, + permits: 20000, strategy: BackpressureStrategy::Drop, }, enable_metrics: false, @@ -99,35 +97,16 @@ impl StreamClientConfig { /// This configuration prioritizes latency over throughput by: /// - Processing events immediately without buffering /// - Implementing a blocking backpressure strategy to ensure no data loss - /// - Setting minimal permits (1) to minimize memory usage + /// - Setting optimal permits (4000) for balanced throughput and latency /// /// Ideal for scenarios where every millisecond counts and you cannot /// afford to lose any events, such as trading applications or real-time monitoring. pub fn low_latency() -> Self { Self { connection: ConnectionConfig::default(), - backpressure: BackpressureConfig { permits: 1, strategy: BackpressureStrategy::Block }, + backpressure: BackpressureConfig { permits: 4000, strategy: BackpressureStrategy::Block }, enable_metrics: false, } } - /// Creates an asynchronous processing configuration optimized for high-volume scenarios. - /// - /// This configuration balances throughput and reliability by: - /// - Implementing an async backpressure strategy for non-blocking operation - /// - Setting a balanced permit buffer (5,000) for steady flow - /// - /// Ideal for scenarios where you need sustained high throughput with - /// fire-and-forget semantics, such as data ingestion pipelines or - /// event streaming applications where some eventual consistency is acceptable. - pub fn async_processing() -> Self { - Self { - connection: ConnectionConfig::default(), - backpressure: BackpressureConfig { - permits: 5000, - strategy: BackpressureStrategy::Async, - }, - enable_metrics: false, - } - } } diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 571a6b5..7ae82f7 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -1,17 +1,19 @@ +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Instant; +use crossbeam_queue::SegQueue; use solana_sdk::pubkey::Pubkey; -use solana_sdk::signature::Signature; -use tokio::sync::Semaphore; use crate::common::AnyResult; +use crate::streaming::common::BackpressureStrategy; use crate::streaming::common::{ MetricsEventType, MetricsManager, StreamClientConfig as ClientConfig, }; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::core::account_event_parser::AccountEventParser; use crate::streaming::event_parser::core::common_event_parser::CommonEventParser; +use crate::streaming::event_parser::core::traits::get_high_perf_clock; use crate::streaming::event_parser::EventParser; use crate::streaming::event_parser::{ core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol, @@ -20,7 +22,7 @@ use crate::streaming::grpc::{BackpressureConfig, EventPretty}; use crate::streaming::shred::TransactionWithSlot; use once_cell::sync::OnceCell; -/// Event processor +/// High-performance Event processor using SegQueue for all strategies pub struct EventProcessor { pub(crate) metrics_manager: MetricsManager, pub(crate) config: ClientConfig, @@ -29,15 +31,27 @@ pub struct EventProcessor { pub(crate) event_type_filter: Option, pub(crate) callback: Option) + Send + Sync>>, pub(crate) backpressure_config: BackpressureConfig, - /// Backpressure semaphore for controlling concurrent processing count - pub(crate) backpressure_semaphore: Arc, + /// High-performance lockfree queue for gRPC events + pub(crate) grpc_queue: Arc)>>, + /// High-performance lockfree queue for shred events + pub(crate) shred_queue: Arc)>>, + /// Fast O(1) counter for Drop strategy (avoids expensive SegQueue::len()) + pub(crate) grpc_pending_count: Arc, + pub(crate) shred_pending_count: Arc, + /// Processing thread control + pub(crate) processing_shutdown: Arc, } impl EventProcessor { - /// Create a new event processor + /// Create a new high-performance event processor pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self { let backpressure_config = config.backpressure.clone(); - let backpressure_semaphore = Arc::new(Semaphore::new(backpressure_config.permits)); + let grpc_queue = Arc::new(SegQueue::new()); + let shred_queue = Arc::new(SegQueue::new()); + let grpc_pending_count = Arc::new(AtomicUsize::new(0)); + let shred_pending_count = Arc::new(AtomicUsize::new(0)); + let processing_shutdown = Arc::new(AtomicBool::new(false)); + Self { metrics_manager, config, @@ -46,7 +60,11 @@ impl EventProcessor { event_type_filter: None, backpressure_config, callback: None, - backpressure_semaphore, + grpc_queue, + shred_queue, + grpc_pending_count, + shred_pending_count, + processing_shutdown, } } @@ -57,94 +75,100 @@ impl EventProcessor { backpressure_config: BackpressureConfig, callback: Option) + Send + Sync>>, ) { - self.protocols = protocols.clone(); - self.event_type_filter = event_type_filter.clone(); - // Recreate semaphore if backpressure configuration changes - if self.backpressure_config.permits != backpressure_config.permits { - self.backpressure_semaphore = Arc::new(Semaphore::new(backpressure_config.permits)); - } + self.protocols = protocols; + self.event_type_filter = event_type_filter; + + // Check if Block processing thread should be started (before moving backpressure_config) + let should_start_block_processing = true; + // matches!(backpressure_config.strategy, BackpressureStrategy::Block); + self.backpressure_config = backpressure_config; self.callback = callback; - self.parser_cache - .get_or_init(|| Arc::new(MutilEventParser::new(protocols, event_type_filter))); + // Use stored values to initialize parser_cache + let protocols_ref = &self.protocols; + let event_type_filter_ref = self.event_type_filter.as_ref(); + self.parser_cache.get_or_init(|| { + Arc::new(MutilEventParser::new(protocols_ref.clone(), event_type_filter_ref.cloned())) + }); + + // Start Block processing thread if using Block strategy + if should_start_block_processing { + self.start_block_processing_thread(); + } } pub fn get_parser(&self) -> Arc { self.parser_cache.get().unwrap().clone() } + /// Create adapter callback + fn create_adapter_callback(&self) -> Arc) + Send + Sync> { + let callback = self.callback.clone().unwrap(); + let metrics_manager = self.metrics_manager.clone(); + + Arc::new(move |event: Box| { + let processing_time_us = event.program_handle_time_consuming_us() as f64; + callback(event); + metrics_manager.update_metrics(MetricsEventType::Transaction, 1, processing_time_us); + }) + } + pub async fn process_grpc_event_transaction_with_metrics( &self, event_pretty: EventPretty, bot_wallet: Option, ) -> AnyResult<()> { - // Backpressure control logic - let backpressure_start = Instant::now(); - let result = self.apply_backpressure_control(event_pretty, bot_wallet).await; - let backpressure_duration = backpressure_start.elapsed(); - - // Record backpressure-related metrics - self.metrics_manager.record_backpressure_metrics( - backpressure_duration, - result.is_ok(), - self.backpressure_semaphore.available_permits(), - ); - - result + self.apply_backpressure_control(event_pretty, bot_wallet).await } - /// Apply backpressure control strategy + /// Apply backpressure control strategy async fn apply_backpressure_control( &self, event_pretty: EventPretty, bot_wallet: Option, ) -> AnyResult<()> { - use crate::streaming::common::BackpressureStrategy; - match self.backpressure_config.strategy { BackpressureStrategy::Block => { - // Blocking strategy: acquire semaphore permit - let _permit = - self.backpressure_semaphore.acquire().await.map_err(|e| { - anyhow::anyhow!("Failed to acquire backpressure permit: {}", e) - })?; - self.process_grpc_event_transaction(event_pretty, bot_wallet).await - } - BackpressureStrategy::Drop => { - // Drop strategy: try to acquire permit, drop if failed - match self.backpressure_semaphore.try_acquire() { - Ok(_permit) => { - let result = - self.process_grpc_event_transaction(event_pretty, bot_wallet).await; - result - } - Err(_) => { - // Record dropped event - self.metrics_manager.increment_dropped_events(); - Ok(()) + // Block strategy: async wait if queue is full (backpressure control) + loop { + let current_pending = self.grpc_pending_count.load(Ordering::Relaxed); + if current_pending < self.backpressure_config.permits { + self.grpc_queue.push((event_pretty, bot_wallet)); + self.grpc_pending_count.fetch_add(1, Ordering::Relaxed); + break; } + // Async yield to avoid blocking gRPC data source + tokio::task::yield_now().await; } - } - BackpressureStrategy::Async => { - // Async strategy: process asynchronously regardless of permits - self.spawn_async_processing(event_pretty, bot_wallet).await; Ok(()) } - } - } - - /// Process event asynchronously (without waiting for semaphore permit) - async fn spawn_async_processing(&self, event_pretty: EventPretty, bot_wallet: Option) { - let processor = self.clone(); - - tokio::spawn(async move { - // Async strategy: no semaphore control, allow unlimited concurrency - // Execute actual event processing directly - if let Err(e) = processor.process_grpc_event_transaction(event_pretty, bot_wallet).await - { - log::error!("Error in async event processing: {}", e); + BackpressureStrategy::Drop => { + // Drop strategy: Use O(1) atomic counter instead of expensive O(n) len() + // If pending count >= permits, DROP the event immediately + let current_pending = self.grpc_pending_count.load(Ordering::Relaxed); + if current_pending >= self.backpressure_config.permits { + self.metrics_manager.increment_dropped_events(); + Ok(()) + } else { + self.grpc_pending_count.fetch_add(1, Ordering::Relaxed); + let processor = self.clone(); + tokio::spawn(async move { + match processor + .process_grpc_event_transaction(event_pretty, bot_wallet) + .await + { + Ok(_) => { + processor.grpc_pending_count.fetch_sub(1, Ordering::Relaxed); + } + Err(e) => { + log::error!("Error in async gRPC processing: {}", e); + } + } + }); + Ok(()) + } } - }); + } } async fn process_grpc_event_transaction( @@ -158,21 +182,15 @@ impl EventProcessor { 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(), + &self.protocols, account_pretty, - self.event_type_filter.clone(), + self.event_type_filter.as_ref(), ); if let Some(event) = account_event { let processing_time_us = event.program_handle_time_consuming_us() as f64; self.invoke_callback(event); - self.update_metrics( - MetricsEventType::Account, - 1, - processing_time_us, - Some(signature), - ); + self.update_metrics(MetricsEventType::Account, 1, processing_time_us); } } EventPretty::Transaction(transaction_pretty) => { @@ -185,18 +203,7 @@ impl EventProcessor { let transaction_index = transaction_pretty.transaction_index; // Use cache to get parser let parser = self.get_parser(); - let callback = self.callback.clone().unwrap(); - let metrics_manager = self.metrics_manager.clone(); - let adapter_callback = Arc::new(move |event: Box| { - let processing_time_us = event.program_handle_time_consuming_us() as f64; - callback(event); - metrics_manager.update_metrics( - MetricsEventType::Transaction, - 1, - processing_time_us, - Some(signature), - ); - }); + let adapter_callback = self.create_adapter_callback(); parser .parse_transaction_owned( tx, @@ -224,7 +231,7 @@ impl EventProcessor { ); let processing_time_us = block_meta_event.program_handle_time_consuming_us() as f64; self.invoke_callback(block_meta_event); - self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us, None); + self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us); } } @@ -253,18 +260,7 @@ impl EventProcessor { bot_wallet: Option, ) -> AnyResult<()> { // Backpressure control logic - let backpressure_start = Instant::now(); - let result = self.apply_shred_backpressure_control(transaction_with_slot, bot_wallet).await; - let backpressure_duration = backpressure_start.elapsed(); - - // Record backpressure-related metrics - self.metrics_manager.record_backpressure_metrics( - backpressure_duration, - result.is_ok(), - self.backpressure_semaphore.available_permits(), - ); - - result + self.apply_shred_backpressure_control(transaction_with_slot, bot_wallet).await } /// Apply shred backpressure control strategy @@ -273,57 +269,47 @@ impl EventProcessor { transaction_with_slot: TransactionWithSlot, bot_wallet: Option, ) -> AnyResult<()> { - use crate::streaming::common::BackpressureStrategy; - match self.backpressure_config.strategy { BackpressureStrategy::Block => { - // Blocking strategy: acquire semaphore permit - let _permit = - self.backpressure_semaphore.acquire().await.map_err(|e| { - anyhow::anyhow!("Failed to acquire backpressure permit: {}", e) - })?; - self.process_shred_transaction(transaction_with_slot, bot_wallet).await - } - BackpressureStrategy::Drop => { - // Drop strategy: try to acquire permit, drop if failed - match self.backpressure_semaphore.try_acquire() { - Ok(_permit) => { - let result = - self.process_shred_transaction(transaction_with_slot, bot_wallet).await; - result - } - Err(_) => { - // Record dropped event - self.metrics_manager.increment_dropped_events(); - Ok(()) + // Block strategy: async wait if queue is full (backpressure control) + loop { + let current_pending = self.shred_pending_count.load(Ordering::Relaxed); + if current_pending < self.backpressure_config.permits { + self.shred_queue.push((transaction_with_slot, bot_wallet)); + self.shred_pending_count.fetch_add(1, Ordering::Relaxed); + break; } + // Async yield to avoid blocking shred data source + tokio::task::yield_now().await; } - } - BackpressureStrategy::Async => { - // Async strategy: process asynchronously regardless of permits - self.spawn_async_shred_processing(transaction_with_slot, bot_wallet).await; Ok(()) } - } - } - - /// Process shred event asynchronously (without waiting for semaphore permit) - async fn spawn_async_shred_processing( - &self, - transaction_with_slot: TransactionWithSlot, - bot_wallet: Option, - ) { - let processor = self.clone(); - - tokio::spawn(async move { - // Async strategy: no semaphore control, allow unlimited concurrency - // Execute actual event processing directly - if let Err(e) = - processor.process_shred_transaction(transaction_with_slot, bot_wallet).await - { - log::error!("Error in async shred event processing: {}", e); + BackpressureStrategy::Drop => { + // Drop strategy: Use O(1) atomic counter instead of expensive O(n) len() + let current_pending = self.shred_pending_count.load(Ordering::Relaxed); + if current_pending >= self.backpressure_config.permits { + self.metrics_manager.increment_dropped_events(); + Ok(()) + } else { + self.shred_pending_count.fetch_add(1, Ordering::Relaxed); + let processor = self.clone(); + tokio::spawn(async move { + match processor + .process_shred_transaction(transaction_with_slot, bot_wallet) + .await + { + Ok(_) => { + processor.shred_pending_count.fetch_sub(1, Ordering::Relaxed); + } + Err(e) => { + log::error!("Error in async shred processing: {}", e); + } + } + }); + Ok(()) + } } - }); + } } pub async fn process_shred_transaction( @@ -342,20 +328,7 @@ impl EventProcessor { let program_received_time_us = transaction_with_slot.program_received_time_us; // Use cache to get parser let parser = self.get_parser(); - let callback = self.callback.clone().unwrap(); - let metrics_manager = self.metrics_manager.clone(); - - let adapter_callback = Arc::new(move |event: Box| { - let processing_time_us = event.program_handle_time_consuming_us() as f64; - callback(event); - metrics_manager.update_metrics( - MetricsEventType::Transaction, - 1, - processing_time_us, - Some(signature), - ); - }); - + let adapter_callback = self.create_adapter_callback(); parser .parse_versioned_transaction_owned( tx, @@ -373,14 +346,70 @@ impl EventProcessor { Ok(()) } - fn update_metrics( - &self, - ty: MetricsEventType, - count: u64, - time_us: f64, - signature: Option, - ) { - self.metrics_manager.update_metrics(ty, count, time_us, signature); + fn update_metrics(&self, ty: MetricsEventType, count: u64, time_us: f64) { + self.metrics_manager.update_metrics(ty, count, time_us); + } + + /// Start dedicated processing threads for all strategies + fn start_block_processing_thread(&self) { + // Reset shutdown flag + self.processing_shutdown.store(false, Ordering::Relaxed); + + let grpc_queue = Arc::clone(&self.grpc_queue); + let shred_queue = Arc::clone(&self.shred_queue); + let grpc_pending_count = Arc::clone(&self.grpc_pending_count); + let shred_pending_count = Arc::clone(&self.shred_pending_count); + let shutdown_flag = Arc::clone(&self.processing_shutdown); + let shutdown_flag_clone = Arc::clone(&self.processing_shutdown); + let processor = self.clone(); + let processor_clone = self.clone(); + // 1. 专用线程 + 2. Busy-wait + 4. 无锁处理 + std::thread::spawn(move || { + // 创建blocking runtime for async processing + let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + while !shutdown_flag.load(Ordering::Relaxed) { + if let Some((event_pretty, bot_wallet)) = grpc_queue.pop() { + // Decrement pending counter when consuming from queue + grpc_pending_count.fetch_sub(1, Ordering::Relaxed); + // Process event in blocking runtime + if let Err(e) = rt.block_on( + processor.process_grpc_event_transaction(event_pretty, bot_wallet), + ) { + println!("Error processing gRPC event: {}", e); + } + } else { + // 2. 优化忙等待: 使用轻量级休眠减少CPU占用 + std::thread::yield_now(); + } + } + }); + + // Shred处理也使用相同的低延迟优化 + std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + + while !shutdown_flag_clone.load(Ordering::Relaxed) { + if let Some((transaction_with_slot, bot_wallet)) = shred_queue.pop() { + // Decrement pending counter when consuming from queue + shred_pending_count.fetch_sub(1, Ordering::Relaxed); + // Process transaction in blocking runtime + if let Err(e) = rt.block_on( + processor_clone + .process_shred_transaction(transaction_with_slot, bot_wallet), + ) { + log::error!("Error processing shred transaction: {}", e); + } + } else { + // 优化忙等待: 使用轻量级休眠减少CPU占用 + std::thread::yield_now(); + } + } + }); + } + + /// Stop processing threads + pub fn stop_processing(&self) { + self.processing_shutdown.store(true, Ordering::Relaxed); } } @@ -395,7 +424,11 @@ impl Clone for EventProcessor { event_type_filter: self.event_type_filter.clone(), backpressure_config: self.backpressure_config.clone(), callback: self.callback.clone(), - backpressure_semaphore: self.backpressure_semaphore.clone(), + grpc_queue: self.grpc_queue.clone(), + shred_queue: self.shred_queue.clone(), + grpc_pending_count: self.grpc_pending_count.clone(), + shred_pending_count: self.shred_pending_count.clone(), + processing_shutdown: self.processing_shutdown.clone(), } } } diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index dd02697..402f909 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -1,11 +1,9 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; -use solana_sdk::signature::Signature; - use super::constants::*; -/// 事件类型枚举 +/// Event type enumeration #[derive(Debug, Clone, Copy)] pub enum EventType { Transaction = 0, @@ -13,7 +11,7 @@ pub enum EventType { BlockMeta = 2, } -/// 兼容性别名 +/// Compatibility alias pub type MetricsEventType = EventType; impl EventType { @@ -30,18 +28,18 @@ impl EventType { } } - // 兼容性常量 + // Compatibility constants pub const TX: EventType = EventType::Transaction; } -/// 高性能原子事件指标 +/// High-performance atomic event metrics #[derive(Debug)] struct AtomicEventMetrics { process_count: AtomicU64, events_processed: AtomicU64, events_in_window: AtomicU64, window_start_nanos: AtomicU64, - events_per_second_bits: AtomicU64, // f64 的位表示 + events_per_second_bits: AtomicU64, // Bit representation of f64 } impl AtomicEventMetrics { @@ -55,20 +53,20 @@ impl AtomicEventMetrics { } } - /// 原子地增加处理计数 + /// Atomically increment process count #[inline] fn add_process_count(&self) { self.process_count.fetch_add(1, Ordering::Relaxed); } - /// 原子地增加事件处理数量 + /// Atomically increment event processing count #[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); } - /// 获取当前计数（非阻塞） + /// Get current count (non-blocking) #[inline] fn get_counts(&self) -> (u64, u64, u64) { ( @@ -78,19 +76,19 @@ impl AtomicEventMetrics { ) } - /// 原子地更新每秒事件数 + /// Atomically update events per second #[inline] fn update_events_per_second(&self, eps: f64) { self.events_per_second_bits.store(eps.to_bits(), Ordering::Relaxed); } - /// 获取每秒事件数 + /// Get events per second #[inline] fn get_events_per_second(&self) -> f64 { f64::from_bits(self.events_per_second_bits.load(Ordering::Relaxed)) } - /// 重置窗口计数 + /// Reset window count #[inline] fn reset_window(&self, new_start_nanos: u64) { self.events_in_window.store(0, Ordering::Relaxed); @@ -103,13 +101,14 @@ impl AtomicEventMetrics { } } -/// 高性能原子处理时间统计 +/// High-performance atomic processing time statistics #[derive(Debug)] struct AtomicProcessingTimeStats { min_time_bits: AtomicU64, max_time_bits: AtomicU64, - max_time_timestamp_nanos: AtomicU64, // 最大值更新时间戳（纳秒） - total_time_us: AtomicU64, // 存储微秒的整数部分 + min_time_timestamp_nanos: AtomicU64, // Timestamp of min value update (nanoseconds) + max_time_timestamp_nanos: AtomicU64, // Timestamp of max value update (nanoseconds) + total_time_us: AtomicU64, // Store integer part of microseconds total_events: AtomicU64, } @@ -122,13 +121,14 @@ impl AtomicProcessingTimeStats { Self { min_time_bits: AtomicU64::new(f64::INFINITY.to_bits()), max_time_bits: AtomicU64::new(0), + min_time_timestamp_nanos: AtomicU64::new(now_nanos), max_time_timestamp_nanos: AtomicU64::new(now_nanos), total_time_us: AtomicU64::new(0), total_events: AtomicU64::new(0), } } - /// 原子地更新处理时间统计 + /// Atomically update processing time statistics #[inline] fn update(&self, time_us: f64, event_count: u64) { let time_bits = time_us.to_bits(); @@ -136,8 +136,20 @@ impl AtomicProcessingTimeStats { std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos() as u64; - // 更新最小值（使用 compare_exchange_weak 循环） + // Update minimum value, check time difference and reset if over 10 seconds let mut current_min = self.min_time_bits.load(Ordering::Relaxed); + let min_timestamp = self.min_time_timestamp_nanos.load(Ordering::Relaxed); + + // Check if min value timestamp exceeds 10 seconds (10_000_000_000 nanoseconds) + let min_time_diff_nanos = now_nanos.saturating_sub(min_timestamp); + if min_time_diff_nanos > 10_000_000_000 { + // Over 10 seconds, reset min value + self.min_time_bits.store(f64::INFINITY.to_bits(), Ordering::Relaxed); + self.min_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); + current_min = f64::INFINITY.to_bits(); + } + + // If current time is less than min value, update min value and timestamp while time_bits < current_min { match self.min_time_bits.compare_exchange_weak( current_min, @@ -145,25 +157,29 @@ impl AtomicProcessingTimeStats { Ordering::Relaxed, Ordering::Relaxed, ) { - Ok(_) => break, + Ok(_) => { + // Successfully updated min value, also update timestamp + self.min_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); + break; + } Err(x) => current_min = x, } } - // 更新最大值，检查时间差并在超过10秒时清零 + // Update maximum value, check time difference and reset if over 10 seconds let mut current_max = self.max_time_bits.load(Ordering::Relaxed); let max_timestamp = self.max_time_timestamp_nanos.load(Ordering::Relaxed); - // 检查最大值的时间戳是否超过10秒（10_000_000_000纳秒） + // Check if max value timestamp exceeds 10 seconds (10_000_000_000 nanoseconds) let time_diff_nanos = now_nanos.saturating_sub(max_timestamp); if time_diff_nanos > 10_000_000_000 { - // 超过10秒，清零最大值 + // Over 10 seconds, reset max value self.max_time_bits.store(0, Ordering::Relaxed); self.max_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); current_max = 0; } - // 如果当前时间大于最大值，更新最大值和时间戳 + // If current time is greater than max value, update max value and timestamp while time_bits > current_max { match self.max_time_bits.compare_exchange_weak( current_max, @@ -172,7 +188,7 @@ impl AtomicProcessingTimeStats { Ordering::Relaxed, ) { Ok(_) => { - // 成功更新最大值，同时更新时间戳 + // Successfully updated max value, also update timestamp self.max_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed); break; } @@ -180,13 +196,13 @@ impl AtomicProcessingTimeStats { } } - // 更新累计值（将微秒转换为整数避免浮点累加问题） + // Update cumulative values (convert microseconds to integers to avoid floating point accumulation issues) 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); } - /// 获取统计值（非阻塞） + /// Get statistics (non-blocking) #[inline] fn get_stats(&self) -> ProcessingTimeStats { let min_bits = self.min_time_bits.load(Ordering::Relaxed); @@ -207,7 +223,7 @@ impl AtomicProcessingTimeStats { } } -/// 处理时间统计结果 +/// Processing time statistics result #[derive(Debug, Clone)] pub struct ProcessingTimeStats { pub min_us: f64, @@ -215,7 +231,7 @@ pub struct ProcessingTimeStats { pub avg_us: f64, } -/// 事件指标快照 +/// Event metrics snapshot #[derive(Debug, Clone)] pub struct EventMetricsSnapshot { pub process_count: u64, @@ -223,19 +239,7 @@ pub struct EventMetricsSnapshot { pub events_per_second: f64, } -/// 背压指标快照 -#[derive(Debug, Clone)] -pub struct BackpressureMetricsSnapshot { - pub total_duration_us: u64, - pub success_count: u64, - pub failure_count: u64, - pub min_permits: u64, - pub max_permits: u64, - pub avg_duration_us: f64, - pub success_rate: f64, -} - -/// 兼容性结构 - 完整的性能指标 +/// Compatibility structure - complete performance metrics #[derive(Debug, Clone)] pub struct PerformanceMetrics { pub uptime: std::time::Duration, @@ -243,25 +247,15 @@ pub struct PerformanceMetrics { pub account_metrics: EventMetricsSnapshot, pub block_meta_metrics: EventMetricsSnapshot, pub processing_stats: ProcessingTimeStats, - pub backpressure_metrics: BackpressureMetricsSnapshot, pub dropped_events_count: u64, } impl PerformanceMetrics { - /// 创建默认的性能指标（兼容性方法） + /// Create default performance metrics (compatibility method) pub fn new() -> Self { let default_metrics = EventMetricsSnapshot { process_count: 0, events_processed: 0, events_per_second: 0.0 }; let default_stats = ProcessingTimeStats { min_us: 0.0, max_us: 0.0, avg_us: 0.0 }; - let default_backpressure = BackpressureMetricsSnapshot { - total_duration_us: 0, - success_count: 0, - failure_count: 0, - min_permits: 0, - max_permits: 0, - avg_duration_us: 0.0, - success_rate: 0.0, - }; Self { uptime: std::time::Duration::ZERO, @@ -269,24 +263,17 @@ impl PerformanceMetrics { account_metrics: default_metrics.clone(), block_meta_metrics: default_metrics, processing_stats: default_stats, - backpressure_metrics: default_backpressure, dropped_events_count: 0, } } } -/// 高性能指标系统 +/// High-performance metrics system #[derive(Debug)] pub struct HighPerformanceMetrics { start_nanos: u64, event_metrics: [AtomicEventMetrics; 3], processing_stats: AtomicProcessingTimeStats, - // 背压相关指标 - backpressure_total_duration_us: AtomicU64, - backpressure_success_count: AtomicU64, - backpressure_failure_count: AtomicU64, - backpressure_min_permits: AtomicU64, - backpressure_max_permits: AtomicU64, // 丢弃事件指标 dropped_events_count: AtomicU64, } @@ -305,12 +292,6 @@ impl HighPerformanceMetrics { AtomicEventMetrics::new(now_nanos), ], processing_stats: AtomicProcessingTimeStats::new(), - // 初始化背压相关指标 - backpressure_total_duration_us: AtomicU64::new(0), - backpressure_success_count: AtomicU64::new(0), - backpressure_failure_count: AtomicU64::new(0), - backpressure_min_permits: AtomicU64::new(u64::MAX), // 初始化为最大值，便于后续比较 - backpressure_max_permits: AtomicU64::new(0), // 初始化丢弃事件指标 dropped_events_count: AtomicU64::new(0), } @@ -341,32 +322,6 @@ impl HighPerformanceMetrics { self.processing_stats.get_stats() } - /// 获取背压指标快照 - #[inline] - pub fn get_backpressure_metrics(&self) -> BackpressureMetricsSnapshot { - let total_duration_us = self.backpressure_total_duration_us.load(Ordering::Relaxed); - let success_count = self.backpressure_success_count.load(Ordering::Relaxed); - let failure_count = self.backpressure_failure_count.load(Ordering::Relaxed); - let min_permits = self.backpressure_min_permits.load(Ordering::Relaxed); - let max_permits = self.backpressure_max_permits.load(Ordering::Relaxed); - - let total_count = success_count + failure_count; - let avg_duration_us = - if total_count > 0 { total_duration_us as f64 / total_count as f64 } else { 0.0 }; - let success_rate = - if total_count > 0 { success_count as f64 / total_count as f64 } else { 0.0 }; - - BackpressureMetricsSnapshot { - total_duration_us, - success_count, - failure_count, - min_permits: if min_permits == u64::MAX { 0 } else { min_permits }, - max_permits, - avg_duration_us, - success_rate, - } - } - /// 获取丢弃事件计数 #[inline] pub fn get_dropped_events_count(&self) -> u64 { @@ -509,19 +464,13 @@ impl MetricsManager { /// 记录慢处理操作 #[inline] - pub fn log_slow_processing( - &self, - processing_time_us: f64, - event_count: usize, - signature: Option, - ) { + 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: {:.2}us for {} events, signature: {:?}", + log::debug!( + "{} slow processing: {:.2}us for {} events", self.stream_name, processing_time_us, event_count, - signature ); } } @@ -541,11 +490,6 @@ impl MetricsManager { self.metrics.get_processing_stats() } - /// 获取背压指标 - pub fn get_backpressure_metrics(&self) -> BackpressureMetricsSnapshot { - self.metrics.get_backpressure_metrics() - } - /// 获取丢弃事件计数 pub fn get_dropped_events_count(&self) -> u64 { self.metrics.get_dropped_events_count() @@ -556,22 +500,6 @@ impl MetricsManager { println!("\n📊 {} Performance Metrics", self.stream_name); println!(" Run Time: {:?}", self.get_uptime()); - // 打印背压指标表格 - let backpressure = self.get_backpressure_metrics(); - if backpressure.success_count > 0 || backpressure.failure_count > 0 { - println!("\n🚦 Backpressure Metrics"); - println!("┌──────────────────────┬─────────────┐"); - println!("│ Metric │ Value │"); - println!("├──────────────────────┼─────────────┤"); - println!("│ Success Count │ {:11} │", backpressure.success_count); - println!("│ Failure Count │ {:11} │", backpressure.failure_count); - println!("│ Success Rate │ {:11.2} │", backpressure.success_rate * 100.0); - println!("│ Avg Duration (ms) │ {:11.2} │", backpressure.avg_duration_us / 1000.0); - println!("│ Min Permits │ {:11} │", backpressure.min_permits); - println!("│ Max Permits │ {:11} │", backpressure.max_permits); - println!("└──────────────────────┴─────────────┘"); - } - // 打印丢弃事件指标 let dropped_count = self.get_dropped_events_count(); if dropped_count > 0 { @@ -603,7 +531,7 @@ impl MetricsManager { println!("│ Metric │ Value (us) │"); println!("├───────────────────────┼─────────────┤"); println!("│ Average │ {:9.2} │", stats.avg_us); - println!("│ Minimum │ {:9.2} │", stats.min_us); + println!("│ Minimum within 10s │ {:9.2} │", stats.min_us); println!("│ Maximum within 10s │ {:9.2} │", stats.max_us); println!("└───────────────────────┴─────────────┘"); @@ -648,7 +576,6 @@ impl MetricsManager { account_metrics: self.get_event_metrics(EventType::Account), block_meta_metrics: self.get_event_metrics(EventType::BlockMeta), processing_stats: self.get_processing_stats(), - backpressure_metrics: self.metrics.get_backpressure_metrics(), dropped_events_count: self.metrics.get_dropped_events_count(), } } @@ -678,76 +605,9 @@ impl MetricsManager { event_type: MetricsEventType, events_processed: u64, processing_time_us: f64, - signature: Option, ) { self.record_events(event_type, events_processed, processing_time_us); - self.log_slow_processing(processing_time_us, events_processed as usize, signature); - } - - /// 记录背压相关的metrics - #[inline] - pub fn record_backpressure_metrics( - &self, - backpressure_duration: std::time::Duration, - success: bool, - available_permits: usize, - ) { - if !self.enable_metrics { - return; - } - - let duration_us = backpressure_duration.as_micros() as u64; - let permits = available_permits as u64; - - // 记录总持续时间 - self.metrics.backpressure_total_duration_us.fetch_add(duration_us, Ordering::Relaxed); - - // 记录成功/失败计数 - if success { - self.metrics.backpressure_success_count.fetch_add(1, Ordering::Relaxed); - } else { - self.metrics.backpressure_failure_count.fetch_add(1, Ordering::Relaxed); - } - - // 更新最小许可数（使用 compare_exchange_weak 循环） - let mut current_min = self.metrics.backpressure_min_permits.load(Ordering::Relaxed); - while permits < current_min { - match self.metrics.backpressure_min_permits.compare_exchange_weak( - current_min, - permits, - Ordering::Relaxed, - Ordering::Relaxed, - ) { - Ok(_) => break, - Err(x) => current_min = x, - } - } - - // 更新最大许可数 - let mut current_max = self.metrics.backpressure_max_permits.load(Ordering::Relaxed); - while permits > current_max { - match self.metrics.backpressure_max_permits.compare_exchange_weak( - current_max, - permits, - Ordering::Relaxed, - Ordering::Relaxed, - ) { - Ok(_) => break, - Err(x) => current_max = x, - } - } - - // 记录慢背压操作的日志 - if duration_us > 10_000 { - // 超过10ms的背压认为是慢操作 - log::warn!( - "{} slow backpressure: {:.2}ms, success: {}, available_permits: {}", - self.stream_name, - duration_us as f64 / 1000.0, - success, - available_permits - ); - } + self.log_slow_processing(processing_time_us, events_processed as usize); } /// 增加丢弃事件计数 @@ -762,7 +622,7 @@ impl MetricsManager { // 每丢弃1000个事件记录一次警告日志 if new_count % 1000 == 0 { - log::warn!("{} dropped events count reached: {}", self.stream_name, new_count); + log::debug!("{} dropped events count reached: {}", self.stream_name, new_count); } } @@ -774,17 +634,22 @@ impl MetricsManager { } // 原子地增加丢弃事件计数 - let new_count = self.metrics.dropped_events_count.fetch_add(count, Ordering::Relaxed) + count; + let new_count = + self.metrics.dropped_events_count.fetch_add(count, Ordering::Relaxed) + count; // 记录批量丢弃事件的日志 if count > 1 { - log::warn!("{} dropped batch of {} events, total dropped: {}", - self.stream_name, count, new_count); + log::debug!( + "{} dropped batch of {} events, total dropped: {}", + self.stream_name, + count, + new_count + ); } // 每丢弃1000个事件记录一次警告日志 if new_count % 1000 == 0 || (new_count / 1000) != ((new_count - count) / 1000) { - log::warn!("{} dropped events count reached: {}", self.stream_name, new_count); + log::debug!("{} dropped events count reached: {}", self.stream_name, new_count); } } } diff --git a/src/streaming/common/mod.rs b/src/streaming/common/mod.rs index 011bab8..55a76aa 100644 --- a/src/streaming/common/mod.rs +++ b/src/streaming/common/mod.rs @@ -1,15 +1,15 @@ // 公用模块 - 包含流处理相关的通用功能 pub mod config; pub mod metrics; -pub mod batch; pub mod constants; pub mod subscription; pub mod event_processor; +pub mod simd_utils; // 重新导出主要类型 pub use config::*; pub use metrics::*; -pub use batch::*; pub use constants::*; pub use subscription::*; -pub use event_processor::*; \ No newline at end of file +pub use event_processor::*; +pub use simd_utils::*; \ No newline at end of file diff --git a/src/streaming/common/simd_utils.rs b/src/streaming/common/simd_utils.rs new file mode 100644 index 0000000..8aaf1f0 --- /dev/null +++ b/src/streaming/common/simd_utils.rs @@ -0,0 +1,295 @@ +use wide::*; + +/// SIMD-accelerated data parsing utilities +pub struct SimdUtils; + +impl SimdUtils { + /// SIMD-accelerated byte array comparison + /// For arrays with length >= 16, uses SIMD instructions for fast comparison + #[inline(always)] + pub fn fast_bytes_equal(a: &[u8], b: &[u8]) -> bool { + if a.len() != b.len() { + return false; + } + + let len = a.len(); + + // For small arrays, use standard comparison directly + if len < 16 { + return a == b; + } + + // Use SIMD to process 16-byte chunks + let chunks = len / 16; + let remainder = len % 16; + + // Process complete 16-byte chunks + for i in 0..chunks { + let offset = i * 16; + let chunk_a = u8x16::from(&a[offset..offset + 16]); + let chunk_b = u8x16::from(&b[offset..offset + 16]); + + if !chunk_a.cmp_eq(chunk_b).all() { + return false; + } + } + + // Process remaining bytes + if remainder > 0 { + let start = chunks * 16; + return &a[start..] == &b[start..]; + } + + true + } + + /// Fast discriminator matching, specifically for instruction discriminator comparison + #[inline(always)] + pub fn fast_discriminator_match(data: &[u8], discriminator: &[u8]) -> bool { + if data.len() < discriminator.len() { + return false; + } + + let disc_len = discriminator.len(); + + // Optimize for common discriminator lengths + match disc_len { + 1 => data[0] == discriminator[0], + 2 => { + let data_u16 = u16::from_le_bytes([data[0], data[1]]); + let disc_u16 = u16::from_le_bytes([discriminator[0], discriminator[1]]); + data_u16 == disc_u16 + } + 4 => { + let data_u32 = u32::from_le_bytes([data[0], data[1], data[2], data[3]]); + let disc_u32 = u32::from_le_bytes([ + discriminator[0], + discriminator[1], + discriminator[2], + discriminator[3], + ]); + data_u32 == disc_u32 + } + 8 => { + let data_u64 = u64::from_le_bytes([ + data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7], + ]); + let disc_u64 = u64::from_le_bytes([ + discriminator[0], + discriminator[1], + discriminator[2], + discriminator[3], + discriminator[4], + discriminator[5], + discriminator[6], + discriminator[7], + ]); + data_u64 == disc_u64 + } + 16 => { + // Use SIMD to process 16-byte discriminators + let data_chunk = u8x16::from(&data[..16]); + let disc_chunk = u8x16::from(discriminator); + data_chunk.cmp_eq(disc_chunk).all() + } + _ => { + // For other lengths, use generic SIMD comparison + Self::fast_bytes_equal(&data[..disc_len], discriminator) + } + } + } + + /// SIMD-accelerated memory search to find specific patterns in data + #[inline(always)] + pub fn find_pattern_simd(haystack: &[u8], needle: &[u8]) -> Option { + if needle.is_empty() || haystack.len() < needle.len() { + return None; + } + + let needle_len = needle.len(); + let haystack_len = haystack.len(); + + // For single-byte search, use optimized method + if needle_len == 1 { + let target = needle[0]; + return haystack.iter().position(|&b| b == target); + } + + // For multi-byte search, use SIMD acceleration + if needle_len <= 16 && haystack_len >= 16 { + let first_byte = needle[0]; + let chunks = (haystack_len - needle_len + 1) / 16; + + for chunk_idx in 0..chunks { + let start = chunk_idx * 16; + let end = std::cmp::min(start + 16, haystack_len - needle_len + 1); + + // Use SIMD to find first byte matches + let chunk = &haystack[start..start + 16]; + let target_vec = u8x16::splat(first_byte); + let chunk_vec = u8x16::from(chunk); + let matches = chunk_vec.cmp_eq(target_vec); + + // Check each match position + let matches_array: [u8; 16] = matches.into(); + for i in 0..16 { + if start + i >= end { + break; + } + + if matches_array[i] != 0 && start + i + needle_len <= haystack_len { + if Self::fast_bytes_equal( + &haystack[start + i..start + i + needle_len], + needle, + ) { + return Some(start + i); + } + } + } + } + + // Process remaining part + let remaining_start = chunks * 16; + for i in remaining_start..=(haystack_len - needle_len) { + if Self::fast_bytes_equal(&haystack[i..i + needle_len], needle) { + return Some(i); + } + } + } else { + // Fallback to standard search + for i in 0..=(haystack_len - needle_len) { + if Self::fast_bytes_equal(&haystack[i..i + needle_len], needle) { + return Some(i); + } + } + } + + None + } + + /// SIMD-accelerated data validation to check if data conforms to specific format + #[inline(always)] + pub fn validate_data_format(data: &[u8], min_length: usize) -> bool { + if data.len() < min_length { + return false; + } + + true + } + + /// Fast checksum calculation (maintains API consistency) + #[inline(always)] + pub fn fast_checksum(data: &[u8]) -> u32 { + // Simplified implementation, directly sum all bytes + data.iter().map(|&b| b as u32).sum() + } + + /// SIMD-accelerated data copy (for large data blocks) + #[inline(always)] + pub fn fast_copy(src: &[u8], dst: &mut [u8]) { + if src.len() != dst.len() { + panic!("Source and destination must have the same length"); + } + + let len = src.len(); + + if len >= 32 { + // Use 32-byte SIMD copy + let chunks = len / 32; + + for i in 0..chunks { + let start = i * 32; + let src_chunk1 = u8x16::from(&src[start..start + 16]); + let src_chunk2 = u8x16::from(&src[start + 16..start + 32]); + + let chunk1_array: [u8; 16] = src_chunk1.into(); + let chunk2_array: [u8; 16] = src_chunk2.into(); + + dst[start..start + 16].copy_from_slice(&chunk1_array); + dst[start + 16..start + 32].copy_from_slice(&chunk2_array); + } + + // Process remaining bytes + let remaining_start = chunks * 32; + dst[remaining_start..].copy_from_slice(&src[remaining_start..]); + } else { + // For small data, use standard copy + dst.copy_from_slice(src); + } + } + + /// SIMD-accelerated account indices validation + /// Validates that all indices in the account index array are less than the total account count + #[inline(always)] + pub fn validate_account_indices_simd(indices: &[u8], account_count: usize) -> bool { + if indices.is_empty() { + return true; + } + + let max_valid_index = account_count as u8; + + // For small arrays, use standard comparison directly + if indices.len() < 16 { + return indices.iter().all(|&idx| idx < max_valid_index); + } + + // Use SIMD for batch loading and comparison + let chunks = indices.len() / 16; + let remainder = indices.len() % 16; + + // Process complete 16-byte chunks + for i in 0..chunks { + let start = i * 16; + let indices_chunk = u8x16::from(&indices[start..start + 16]); + + // Convert SIMD vector to array for fast batch checking + let indices_array: [u8; 16] = indices_chunk.into(); + + // Use unrolled loop for fast comparison, compiler will optimize this + if indices_array[0] >= max_valid_index + || indices_array[1] >= max_valid_index + || indices_array[2] >= max_valid_index + || indices_array[3] >= max_valid_index + || indices_array[4] >= max_valid_index + || indices_array[5] >= max_valid_index + || indices_array[6] >= max_valid_index + || indices_array[7] >= max_valid_index + || indices_array[8] >= max_valid_index + || indices_array[9] >= max_valid_index + || indices_array[10] >= max_valid_index + || indices_array[11] >= max_valid_index + || indices_array[12] >= max_valid_index + || indices_array[13] >= max_valid_index + || indices_array[14] >= max_valid_index + || indices_array[15] >= max_valid_index + { + return false; + } + } + + // Process remaining bytes + if remainder > 0 { + let remaining_start = chunks * 16; + return indices[remaining_start..].iter().all(|&idx| idx < max_valid_index); + } + + true + } + + /// SIMD-accelerated instruction data validation + /// Validates basic format and length requirements of instruction data + #[inline(always)] + pub fn validate_instruction_data_simd( + data: &[u8], + min_length: usize, + discriminator_length: usize, + ) -> bool { + // Basic length check + if data.len() < min_length || data.len() < discriminator_length { + return false; + } + + // Use existing data format validation + Self::validate_data_format(data, min_length) + } +} diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index c74b304..3e9d088 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -2,22 +2,12 @@ pub mod types; pub mod utils; pub mod filter; -pub const EMPTY_ID: &str = ""; - /// 自动生成UnifiedEvent trait实现的宏 #[macro_export] macro_rules! impl_unified_event { // 带有自定义ID表达式的版本 ($struct_name:ident, $($field:ident),*) => { impl $crate::streaming::event_parser::core::traits::UnifiedEvent for $struct_name { - fn id(&self) -> &str { - &self.metadata.id - } - - fn clear_id(&mut self) { - self.metadata.id = $crate::streaming::event_parser::common::EMPTY_ID.to_string(); - } - fn event_type(&self) -> $crate::streaming::event_parser::common::types::EventType { self.metadata.event_type.clone() } @@ -66,6 +56,10 @@ macro_rules! impl_unified_event { self.metadata.set_swap_data(swap_data); } + fn swap_data_is_parsed(&self) -> bool { + self.metadata.swap_data.is_some() + } + fn instruction_outer_index(&self) -> i64 { self.metadata.instruction_outer_index } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 60c1e1a..b086d8f 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -2,26 +2,23 @@ use borsh::{BorshDeserialize, BorshSerialize}; use crossbeam_queue::ArrayQueue; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; -use solana_transaction_status::{InnerInstruction, UiInstruction}; -use std::{ - fmt, - hash::{DefaultHasher, Hash, Hasher}, - str::FromStr, - sync::Arc, -}; +use std::{borrow::Cow, fmt, str::FromStr, sync::Arc}; use crate::{ match_event, - streaming::event_parser::{ - protocols::{ - bonk::BonkTradeEvent, - pumpfun::PumpFunTradeEvent, - pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}, - raydium_amm_v4::RaydiumAmmV4SwapEvent, - raydium_clmm::{RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event}, - raydium_cpmm::RaydiumCpmmSwapEvent, + streaming::{ + common::SimdUtils, + event_parser::{ + protocols::{ + bonk::BonkTradeEvent, + pumpfun::PumpFunTradeEvent, + pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}, + raydium_amm_v4::RaydiumAmmV4SwapEvent, + raydium_clmm::{RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event}, + raydium_cpmm::RaydiumCpmmSwapEvent, + }, + UnifiedEvent, }, - UnifiedEvent, }, }; @@ -293,8 +290,7 @@ pub struct SwapData { Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, )] pub struct EventMetadata { - pub id: String, - pub signature: String, + pub signature: Cow<'static, str>, pub slot: u64, pub transaction_index: Option, // 新增：交易在slot中的索引 pub block_time: i64, @@ -312,8 +308,7 @@ pub struct EventMetadata { impl EventMetadata { #[allow(clippy::too_many_arguments)] pub fn new( - id: String, - signature: String, + signature: Cow<'static, str>, slot: u64, block_time: i64, block_time_ms: i64, @@ -326,7 +321,6 @@ impl EventMetadata { transaction_index: Option, ) -> Self { Self { - id, signature, slot, block_time, @@ -343,10 +337,6 @@ impl EventMetadata { } } - pub fn set_id(&mut self, id: String) { - self.id = format!("{}-{}-{}", self.signature, self.event_type, id); - } - pub fn set_swap_data(&mut self, swap_data: SwapData) { self.swap_data = Some(swap_data); } @@ -463,6 +453,12 @@ pub fn parse_swap_data_from_next_instructions( break; } let data = &compiled.data; + + // 使用 SIMD 验证数据格式 + if !SimdUtils::validate_data_format(data, 8) { + continue; + } + let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize]; let (source, destination, amount) = match data[0] { 12 if compiled.accounts.len() >= 4 => { diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 030266a..0efa7be 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -1,11 +1,13 @@ +use std::borrow::Cow; use std::collections::HashMap; use std::sync::OnceLock; use solana_sdk::pubkey::Pubkey; +use crate::streaming::common::SimdUtils; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; -use crate::streaming::event_parser::core::traits::UnifiedEvent; +use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock}; use crate::streaming::event_parser::protocols::bonk::parser::BONK_PROGRAM_ID; use crate::streaming::event_parser::protocols::pumpfun::parser::PUMPFUN_PROGRAM_ID; use crate::streaming::event_parser::protocols::pumpswap::parser::PUMPSWAP_PROGRAM_ID; @@ -35,7 +37,10 @@ static PROTOCOL_CONFIGS_CACHE: OnceLock, event_type_filter: Option) -> Vec { + pub fn configs( + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + ) -> Vec { let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| { let mut map: HashMap> = HashMap::new(); map.insert(Protocol::PumpSwap, vec![ @@ -145,32 +150,39 @@ impl AccountEventParser { }); let mut configs = vec![]; + let empty_vec = vec![]; for protocol in protocols { - let protocol_configs = protocols_map.get(&protocol).unwrap_or(&vec![]).clone(); - let filtered_configs: Vec = protocol_configs.into_iter().filter(|config| { - event_type_filter.as_ref().map(|filter| filter.include.contains(&config.event_type)).unwrap_or(true) - }).collect(); + let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec); + let filtered_configs: Vec = protocol_configs + .iter() + .filter(|config| { + event_type_filter + .map(|filter| filter.include.contains(&config.event_type)) + .unwrap_or(true) + }) + .cloned() + .collect(); configs.extend(filtered_configs); } configs } pub fn parse_account_event( - protocols: Vec, + protocols: &[Protocol], account: AccountPretty, - event_type_filter: Option, + event_type_filter: Option<&EventTypeFilter>, ) -> Option> { let configs = Self::configs(protocols, event_type_filter); for config in configs { if account.owner == config.program_id - && account.data[..config.account_discriminator.len()] - == *config.account_discriminator + && SimdUtils::fast_discriminator_match(&account.data, config.account_discriminator) { + let signature_str = Cow::Owned(account.signature.to_string()); let event = (config.account_parser)( &account, EventMetadata { slot: account.slot, - signature: account.signature.to_string(), + signature: signature_str, protocol: config.protocol_type, event_type: config.event_type, program_id: config.program_id, @@ -180,7 +192,7 @@ impl AccountEventParser { ); if let Some(mut event) = event { event.set_program_handle_time_consuming_us( - chrono::Utc::now().timestamp_micros() - account.program_received_time_us, + get_high_perf_clock().elapsed_micros_since(account.program_received_time_us), ); return Some(event); } diff --git a/src/streaming/event_parser/core/common_event_parser.rs b/src/streaming/event_parser/core/common_event_parser.rs index 6ec37ae..ef44d96 100644 --- a/src/streaming/event_parser/core/common_event_parser.rs +++ b/src/streaming/event_parser/core/common_event_parser.rs @@ -1,4 +1,4 @@ -use crate::streaming::event_parser::core::traits::UnifiedEvent; +use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock}; use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent; pub struct CommonEventParser {} @@ -17,7 +17,7 @@ impl CommonEventParser { program_received_time_us, ); block_meta_event.set_program_handle_time_consuming_us( - chrono::Utc::now().timestamp_micros() - program_received_time_us, + get_high_perf_clock().elapsed_micros_since(program_received_time_us), ); Box::new(block_meta_event) } diff --git a/src/streaming/event_parser/core/global_state.rs b/src/streaming/event_parser/core/global_state.rs index 5e75c0f..c14142e 100644 --- a/src/streaming/event_parser/core/global_state.rs +++ b/src/streaming/event_parser/core/global_state.rs @@ -1,92 +1,174 @@ use solana_sdk::pubkey::Pubkey; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use dashmap::DashMap; +use std::collections::BTreeSet; -/// Global state management, thread-safe implementation without locks +const MAX_SLOTS: usize = 1000; +const CLEANUP_BATCH_SIZE: usize = 100; + +/// Slot-based trader addresses, completely lock-free +#[derive(Default)] +struct SlotAddresses { + /// Developer addresses for this slot + dev_addresses: BTreeSet, + /// Bonk developer addresses for this slot + bonk_dev_addresses: BTreeSet, +} + +/// High-performance global state with lock-free slot-based storage pub struct GlobalState { - /// Last processed slot - last_slot: AtomicU64, - /// Developer address array - dev_addresses: parking_lot::RwLock>, - /// Bonk developer address array - bonk_dev_addresses: parking_lot::RwLock>, + /// Slot -> trader addresses mapping (lock-free concurrent hashmap) + slot_data: DashMap, + /// Current slot count for capacity management + slot_count: AtomicUsize, + /// Generation counter to handle cleanup races + generation: AtomicU64, } impl GlobalState { - /// Create a new global state instance + /// Create a new high-performance global state instance pub fn new() -> Self { Self { - last_slot: AtomicU64::new(0), - dev_addresses: parking_lot::RwLock::new(Vec::new()), - bonk_dev_addresses: parking_lot::RwLock::new(Vec::new()), + slot_data: DashMap::new(), + slot_count: AtomicUsize::new(0), + generation: AtomicU64::new(0), } } - /// Get current slot - pub fn get_last_slot(&self) -> u64 { - self.last_slot.load(Ordering::Relaxed) - } + /// Lock-free capacity management - cleanup old slots when limit exceeded + fn maybe_cleanup(&self) { + let current_count = self.slot_count.load(Ordering::Relaxed); + if current_count <= MAX_SLOTS { + return; + } - /// Update slot, clear arrays if slot changes - pub fn update_slot(&self, new_slot: u64) { - let old_slot = self.last_slot.swap(new_slot, Ordering::Relaxed); + // Use CAS to ensure only one thread performs cleanup + let gen = self.generation.load(Ordering::Relaxed); + if self.generation.compare_exchange_weak(gen, gen + 1, Ordering::Acquire, Ordering::Relaxed).is_err() { + return; // Another thread is cleaning up + } - if old_slot != new_slot { - // Clear arrays when slot changes - let mut dev_addresses = self.dev_addresses.write(); - let mut bonk_dev_addresses = self.bonk_dev_addresses.write(); + // Collect oldest slots (BTreeMap naturally orders by key) + let mut slots_to_remove: Vec = self.slot_data.iter() + .map(|entry| *entry.key()) + .collect(); + + if slots_to_remove.len() <= MAX_SLOTS { + return; // Race condition, already cleaned up + } + + slots_to_remove.sort_unstable(); + slots_to_remove.truncate(CLEANUP_BATCH_SIZE); - dev_addresses.clear(); - bonk_dev_addresses.clear(); + // Remove old slots atomically + for slot in slots_to_remove { + self.slot_data.remove(&slot); + self.slot_count.fetch_sub(1, Ordering::Relaxed); } } - /// Add developer address - pub fn add_dev_address(&self, address: Pubkey) { - let mut dev_addresses = self.dev_addresses.write(); - if !dev_addresses.contains(&address) { - dev_addresses.push(address); - } + /// Add developer address for a specific slot (lock-free) + pub fn add_dev_address(&self, slot: u64, address: Pubkey) { + self.maybe_cleanup(); + + self.slot_data.entry(slot) + .and_modify(|addresses| { + addresses.dev_addresses.insert(address); + }) + .or_insert_with(|| { + self.slot_count.fetch_add(1, Ordering::Relaxed); + let mut slot_addr = SlotAddresses::default(); + slot_addr.dev_addresses.insert(address); + slot_addr + }); } - /// Check if address is a developer address + /// Add Bonk developer address for a specific slot (lock-free) + pub fn add_bonk_dev_address(&self, slot: u64, address: Pubkey) { + self.maybe_cleanup(); + + self.slot_data.entry(slot) + .and_modify(|addresses| { + addresses.bonk_dev_addresses.insert(address); + }) + .or_insert_with(|| { + self.slot_count.fetch_add(1, Ordering::Relaxed); + let mut slot_addr = SlotAddresses::default(); + slot_addr.bonk_dev_addresses.insert(address); + slot_addr + }); + } + + /// High-performance: Check if address is a developer address in specific slot (O(log m)) + pub fn is_dev_address_in_slot(&self, slot: u64, address: &Pubkey) -> bool { + self.slot_data.get(&slot) + .map(|entry| entry.dev_addresses.contains(address)) + .unwrap_or(false) + } + + /// High-performance: Check if address is a Bonk developer address in specific slot (O(log m)) + pub fn is_bonk_dev_address_in_slot(&self, slot: u64, address: &Pubkey) -> bool { + self.slot_data.get(&slot) + .map(|entry| entry.bonk_dev_addresses.contains(address)) + .unwrap_or(false) + } + + /// Check if address is a developer address in any slot (lock-free scan, slower) pub fn is_dev_address(&self, address: &Pubkey) -> bool { - let dev_addresses = self.dev_addresses.read(); - dev_addresses.contains(address) + self.slot_data.iter().any(|entry| entry.dev_addresses.contains(address)) } - /// Add Bonk developer address - pub fn add_bonk_dev_address(&self, address: Pubkey) { - let mut bonk_dev_addresses = self.bonk_dev_addresses.write(); - if !bonk_dev_addresses.contains(&address) { - bonk_dev_addresses.push(address); - } - } - - /// Check if address is a Bonk developer address + /// Check if address is a Bonk developer address in any slot (lock-free scan, slower) pub fn is_bonk_dev_address(&self, address: &Pubkey) -> bool { - let bonk_dev_addresses = self.bonk_dev_addresses.read(); - bonk_dev_addresses.contains(address) + self.slot_data.iter().any(|entry| entry.bonk_dev_addresses.contains(address)) } - /// Get all developer addresses + /// Get all developer addresses from all slots (lock-free aggregation) pub fn get_dev_addresses(&self) -> Vec { - let dev_addresses = self.dev_addresses.read(); - dev_addresses.clone() + let mut all_addresses = BTreeSet::new(); + for entry in self.slot_data.iter() { + for addr in &entry.dev_addresses { + all_addresses.insert(*addr); + } + } + all_addresses.into_iter().collect() } - /// Get all Bonk developer addresses + /// Get all Bonk developer addresses from all slots (lock-free aggregation) pub fn get_bonk_dev_addresses(&self) -> Vec { - let bonk_dev_addresses = self.bonk_dev_addresses.read(); - bonk_dev_addresses.clone() + let mut all_addresses = BTreeSet::new(); + for entry in self.slot_data.iter() { + for addr in &entry.bonk_dev_addresses { + all_addresses.insert(*addr); + } + } + all_addresses.into_iter().collect() } - /// Clear all data - pub fn clear_all_data(&self) { - let mut dev_addresses = self.dev_addresses.write(); - let mut bonk_dev_addresses = self.bonk_dev_addresses.write(); + /// Get developer addresses for a specific slot + pub fn get_dev_addresses_for_slot(&self, slot: u64) -> Vec { + self.slot_data.get(&slot) + .map(|entry| entry.dev_addresses.iter().copied().collect()) + .unwrap_or_default() + } - dev_addresses.clear(); - bonk_dev_addresses.clear(); + /// Get Bonk developer addresses for a specific slot + pub fn get_bonk_dev_addresses_for_slot(&self, slot: u64) -> Vec { + self.slot_data.get(&slot) + .map(|entry| entry.bonk_dev_addresses.iter().copied().collect()) + .unwrap_or_default() + } + + /// Get current slot count + pub fn get_slot_count(&self) -> usize { + self.slot_count.load(Ordering::Relaxed) + } + + /// Clear all data (lock-free) + pub fn clear_all_data(&self) { + self.slot_data.clear(); + self.slot_count.store(0, Ordering::Relaxed); + self.generation.store(0, Ordering::Relaxed); } } @@ -105,14 +187,9 @@ pub fn get_global_state() -> &'static GlobalState { &GLOBAL_STATE } -/// Convenience function: Update slot -pub fn update_slot(slot: u64) { - get_global_state().update_slot(slot); -} - -/// Convenience function: Add developer address -pub fn add_dev_address(address: Pubkey) { - get_global_state().add_dev_address(address); +/// Convenience function: Add developer address for a specific slot +pub fn add_dev_address(slot: u64, address: Pubkey) { + get_global_state().add_dev_address(slot, address); } /// Convenience function: Check if address is a developer address @@ -120,9 +197,9 @@ pub fn is_dev_address(address: &Pubkey) -> bool { get_global_state().is_dev_address(address) } -/// Convenience function: Add Bonk developer address -pub fn add_bonk_dev_address(address: Pubkey) { - get_global_state().add_bonk_dev_address(address); +/// Convenience function: Add Bonk developer address for a specific slot +pub fn add_bonk_dev_address(slot: u64, address: Pubkey) { + get_global_state().add_bonk_dev_address(slot, address); } /// Convenience function: Check if address is a Bonk developer address @@ -139,3 +216,28 @@ pub fn get_dev_addresses() -> Vec { pub fn get_bonk_dev_addresses() -> Vec { get_global_state().get_bonk_dev_addresses() } + +/// Convenience function: Get developer addresses for a specific slot +pub fn get_dev_addresses_for_slot(slot: u64) -> Vec { + get_global_state().get_dev_addresses_for_slot(slot) +} + +/// Convenience function: Get Bonk developer addresses for a specific slot +pub fn get_bonk_dev_addresses_for_slot(slot: u64) -> Vec { + get_global_state().get_bonk_dev_addresses_for_slot(slot) +} + +/// Convenience function: Get current slot count +pub fn get_slot_count() -> usize { + get_global_state().get_slot_count() +} + +/// High-performance: Check if address is a developer address in specific slot +pub fn is_dev_address_in_slot(slot: u64, address: &Pubkey) -> bool { + get_global_state().is_dev_address_in_slot(slot, address) +} + +/// High-performance: Check if address is a Bonk developer address in specific slot +pub fn is_bonk_dev_address_in_slot(slot: u64, address: &Pubkey) -> bool { + get_global_state().is_bonk_dev_address_in_slot(slot, address) +} diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index db74152..65c93de 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -8,33 +8,141 @@ use solana_transaction_status::{ EncodedConfirmedTransactionWithStatusMeta, InnerInstruction, InnerInstructions, TransactionWithStatusMeta, UiInstruction, }; +use std::borrow::Cow; use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; +use std::time::Instant; -use super::global_state::{add_dev_address, is_dev_address, update_slot}; - -use crate::streaming::event_parser::common::{parse_swap_data_from_next_instructions, SwapData}; -use crate::streaming::event_parser::core::global_state::{ - add_bonk_dev_address, is_bonk_dev_address, +use super::global_state::{ + add_bonk_dev_address, add_dev_address, is_bonk_dev_address, is_dev_address, }; + +use crate::streaming::common::simd_utils::SimdUtils; +use crate::streaming::event_parser::common::{parse_swap_data_from_next_instructions, SwapData}; use crate::streaming::event_parser::protocols::pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent}; use crate::streaming::event_parser::{ - common::{utils::*, EventMetadata, EventType, ProtocolType}, + common::{EventMetadata, EventType, ProtocolType}, protocols::{ bonk::{BonkPoolCreateEvent, BonkTradeEvent}, pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, }, }; +/// 高性能时钟管理器，减少系统调用开销 +#[derive(Debug)] +pub struct HighPerformanceClock { + /// 基准时间点（程序启动时的单调时钟时间） + base_instant: Instant, + /// 基准时间点对应的UTC时间戳（微秒） + base_timestamp_us: i64, +} + +impl HighPerformanceClock { + /// 创建新的高性能时钟 + pub fn new() -> Self { + let base_instant = Instant::now(); + let base_timestamp_us = chrono::Utc::now().timestamp_micros(); + + Self { base_instant, base_timestamp_us } + } + + /// 获取当前时间戳（微秒），使用单调时钟计算，避免系统调用 + #[inline(always)] + pub fn now_micros(&self) -> i64 { + let elapsed = self.base_instant.elapsed(); + self.base_timestamp_us + elapsed.as_micros() as i64 + } + + /// 计算从指定时间戳到现在的消耗时间（微秒） + #[inline(always)] + pub fn elapsed_micros_since(&self, start_timestamp_us: i64) -> i64 { + self.now_micros() - start_timestamp_us + } +} + +impl Default for HighPerformanceClock { + fn default() -> Self { + Self::new() + } +} + +/// 全局高性能时钟实例（使用OnceCell避免重复初始化） +static HIGH_PERF_CLOCK: once_cell::sync::OnceCell = + once_cell::sync::OnceCell::new(); + +/// 获取全局高性能时钟实例 +#[inline(always)] +pub fn get_high_perf_clock() -> &'static HighPerformanceClock { + HIGH_PERF_CLOCK.get_or_init(HighPerformanceClock::new) +} + +/// 轻量级事件包装器，避免频繁的Box分配 +#[derive(Debug)] +pub struct EventWrapper { + pub event: T, +} + +impl EventWrapper { + #[inline] + pub fn new(event: T) -> Self { + Self { event } + } + + #[inline] + pub fn into_boxed(self) -> Box { + Box::new(self.event) + } +} + +/// 高性能账户公钥缓存，避免重复Vec分配 +#[derive(Debug)] +pub struct AccountPubkeyCache { + /// 预分配的账户公钥向量，避免每次重新分配 + cache: Vec, +} + +impl AccountPubkeyCache { + /// 创建新的账户公钥缓存 + pub fn new() -> Self { + Self { + cache: Vec::with_capacity(32), // 预分配32个位置，覆盖大多数交易 + } + } + + /// 从指令账户索引构建账户公钥向量，重用缓存内存 + #[inline] + pub fn build_account_pubkeys( + &mut self, + instruction_accounts: &[u8], + all_accounts: &[Pubkey], + ) -> &[Pubkey] { + self.cache.clear(); + + // 确保容量足够，避免动态扩容 + if self.cache.capacity() < instruction_accounts.len() { + self.cache.reserve(instruction_accounts.len() - self.cache.capacity()); + } + + // 快速填充账户公钥 + for &idx in instruction_accounts.iter() { + if (idx as usize) < all_accounts.len() { + self.cache.push(all_accounts[idx as usize]); + } + } + + &self.cache + } +} + +impl Default for AccountPubkeyCache { + fn default() -> Self { + Self::new() + } +} + /// Unified Event Interface - All protocol events must implement this trait pub trait UnifiedEvent: Debug + Send + Sync { - /// Get event ID - fn id(&self) -> &str; - - /// Set event ID - fn clear_id(&mut self); - /// Get event type fn event_type(&self) -> EventType; @@ -70,6 +178,9 @@ pub trait UnifiedEvent: Debug + Send + Sync { /// Set swap data fn set_swap_data(&mut self, swap_data: SwapData); + /// swap_data is parsed + fn swap_data_is_parsed(&self) -> bool; + /// Get index fn instruction_outer_index(&self) -> i64; fn instruction_inner_index(&self) -> Option; @@ -343,7 +454,6 @@ pub trait EventParser: Send + Sync { let versioned_tx = match transaction.transaction.transaction.decode() { Some(tx) => tx, None => { - println!("Failed to decode transaction"); return Ok(()); } }; @@ -363,6 +473,7 @@ pub trait EventParser: Send + Sync { if let UiInstruction::Compiled(ui_compiled) = ui_instruction { // 解码base58编码的data if let Ok(data) = bs58::decode(&ui_compiled.data).into_vec() { + // base64解码 let compiled_instruction = CompiledInstruction { program_id_index: ui_compiled.program_id_index, accounts: ui_compiled.accounts.clone(), @@ -565,6 +676,8 @@ pub struct GenericEventParser { pub program_ids: Vec, // pub inner_instruction_configs: HashMap, Vec>, pub instruction_configs: HashMap, Vec>, + /// 账户公钥缓存，避免重复分配 + pub account_cache: parking_lot::Mutex, } impl GenericEventParser { @@ -580,7 +693,10 @@ impl GenericEventParser { .push(config.clone()); } - Self { program_ids, instruction_configs } + // 初始化账户缓存 + let account_cache = parking_lot::Mutex::new(AccountPubkeyCache::new()); + + Self { program_ids, instruction_configs, account_cache } } /// 通用的内联指令解析方法 @@ -598,11 +714,11 @@ impl GenericEventParser { transaction_index: Option, ) -> Option> { if let Some(parser) = config.inner_instruction_parser { + let signature_str = Cow::Owned(signature.to_string()); let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( - signature.to_string(), - signature.to_string(), + signature_str, slot, timestamp.seconds, block_time_ms, @@ -636,11 +752,11 @@ impl GenericEventParser { transaction_index: Option, ) -> Option> { if let Some(parser) = config.instruction_parser { + let signature_str = Cow::Owned(signature.to_string()); let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; let metadata = EventMetadata::new( - signature.to_string(), - signature.to_string(), + signature_str, slot, timestamp.seconds, block_time_ms, @@ -678,7 +794,8 @@ impl EventParser for GenericEventParser { transaction_index: Option, config: &GenericEventParseConfig, ) -> Vec> { - if inner_instruction.data.len() < 16 { + // Use SIMD-optimized data validation + if !SimdUtils::validate_instruction_data_simd(&inner_instruction.data, 16, 0) { return Vec::new(); } let data = &inner_instruction.data[16..]; @@ -720,80 +837,115 @@ impl EventParser for GenericEventParser { if !self.should_handle(&program_id) { return Ok(()); } - for (disc, configs) in &self.instruction_configs { - if instruction.data.len() < disc.len() { - continue; - } - let discriminator = &instruction.data[..disc.len()]; - let data = &instruction.data[disc.len()..]; - if discriminator == disc { - // 验证账户索引 - if !validate_account_indices(&instruction.accounts, accounts.len()) { - continue; - } - let account_pubkeys: Vec = - instruction.accounts.iter().map(|&idx| accounts[idx as usize]).collect(); - for config in configs { - if config.program_id != program_id { - continue; - } - if let Some(mut event) = self.parse_instruction_event( - config, - data, - &account_pubkeys, - signature, - slot, - block_time, - program_received_time_us, - outer_index, - inner_index, - transaction_index, - ) { - let mut inner_instruction_event: Option> = None; - if inner_instructions.is_some() { - // 解析对应的内部 log 执行 - for inner_instruction in inner_instructions.unwrap().instructions.iter() - { - let result = self.parse_events_from_inner_instruction( - &inner_instruction.instruction, - signature, - slot, - block_time, - program_received_time_us, - outer_index, - inner_index, - transaction_index, - config, - ); - if result.len() > 0 { - inner_instruction_event = Some(result[0].clone()); - } - // 解析swap数据 - let swap_data = parse_swap_data_from_next_instructions( - &*event, - inner_instructions.unwrap(), - inner_index.unwrap_or(-1_i64) as i8, - &accounts, - ); - if let Some(swap_data) = swap_data { - event.set_swap_data(swap_data); - } + // 一维化并行处理：将所有 (discriminator, config) 组合展开并行处理 + let all_processing_params: Vec<_> = self + .instruction_configs + .iter() + .filter(|(disc, _)| { + // Use SIMD-optimized data validation and discriminator matching + SimdUtils::validate_instruction_data_simd(&instruction.data, disc.len(), disc.len()) + && SimdUtils::fast_discriminator_match(&instruction.data, disc) + }) + .flat_map(|(disc, configs)| { + configs + .iter() + .filter(|config| config.program_id == program_id) + .map(move |config| (disc, config)) + }) + .collect(); + + // Use SIMD-optimized account indices validation (只需检查一次) + if !SimdUtils::validate_account_indices_simd(&instruction.accounts, accounts.len()) { + return Ok(()); + } + + // 使用缓存构建账户公钥列表，避免重复分配 (只需构建一次) + let account_pubkeys = { + let mut cache_guard = self.account_cache.lock(); + cache_guard.build_account_pubkeys(&instruction.accounts, accounts).to_vec() + }; + + // 并行处理所有 (discriminator, config) 组合 + let all_results: Vec<_> = all_processing_params + .iter() + .filter_map(|(disc, config)| { + let data = &instruction.data[disc.len()..]; + self.parse_instruction_event( + config, + data, + &account_pubkeys, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + transaction_index, + ) + .map(|event| ((*disc).clone(), (*config).clone(), event)) + }) + .collect(); + + for (_disc, config, mut event) in all_results { + // 阻塞处理：原有的同步逻辑 + let mut inner_instruction_event: Option> = None; + if inner_instructions.is_some() { + let inner_instructions_ref = inner_instructions.unwrap(); + + // 并行执行两个任务 + let (inner_event_result, swap_data_result) = std::thread::scope(|s| { + let inner_event_handle = s.spawn(|| { + for inner_instruction in inner_instructions_ref.instructions.iter() { + let result = self.parse_events_from_inner_instruction( + &inner_instruction.instruction, + signature, + slot, + block_time, + program_received_time_us, + outer_index, + inner_index, + transaction_index, + &config, + ); + if result.len() > 0 { + return Some(result[0].clone()); } } - // 合并事件 - if let Some(inner_instruction_event) = inner_instruction_event { - event.merge(&*inner_instruction_event); + None + }); + + let swap_data_handle = s.spawn(|| { + if !event.swap_data_is_parsed() { + parse_swap_data_from_next_instructions( + &*event, + inner_instructions_ref, + inner_index.unwrap_or(-1_i64) as i8, + &accounts, + ) + } else { + None } - // 设置处理时间 - event.set_program_handle_time_consuming_us( - chrono::Utc::now().timestamp_micros() - program_received_time_us, - ); - event = process_event(event, bot_wallet); - callback(&event); - break; - } + }); + + // 等待两个任务完成 + (inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap()) + }); + + inner_instruction_event = inner_event_result; + if let Some(swap_data) = swap_data_result { + event.set_swap_data(swap_data); } } + // 合并事件 + if let Some(inner_instruction_event) = inner_instruction_event { + event.merge(&*inner_instruction_event); + } + // 设置处理时间（使用高性能时钟） + event.set_program_handle_time_consuming_us( + get_high_perf_clock().elapsed_micros_since(program_received_time_us), + ); + event = process_event(event, bot_wallet); + callback(&event); } Ok(()) } @@ -811,11 +963,11 @@ fn process_event( mut event: Box, bot_wallet: Option, ) -> Box { - update_slot(event.slot()); + let slot = event.slot(); if let Some(token_info) = event.as_any().downcast_ref::() { - add_dev_address(token_info.user); + add_dev_address(slot, token_info.user); if token_info.creator != Pubkey::default() && token_info.creator != token_info.user { - add_dev_address(token_info.creator); + add_dev_address(slot, token_info.creator); } } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { if is_dev_address(&trade_info.user) || is_dev_address(&trade_info.creator) { @@ -844,7 +996,7 @@ fn process_event( trade_info.user_quote_amount_out; } } else if let Some(pool_info) = event.as_any().downcast_ref::() { - add_bonk_dev_address(pool_info.creator); + add_bonk_dev_address(slot, pool_info.creator); } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { if is_bonk_dev_address(&trade_info.payer) { trade_info.is_dev_create_token_trade = true; @@ -854,6 +1006,5 @@ fn process_event( trade_info.is_dev_create_token_trade = false; } } - event.clear_id(); event } diff --git a/src/streaming/event_parser/protocols/block/block_meta_event.rs b/src/streaming/event_parser/protocols/block/block_meta_event.rs index d204eef..8fe4051 100644 --- a/src/streaming/event_parser/protocols/block/block_meta_event.rs +++ b/src/streaming/event_parser/protocols/block/block_meta_event.rs @@ -1,3 +1,5 @@ +use std::borrow::Cow; + use crate::impl_unified_event; use crate::streaming::event_parser::common::{types::EventType, EventMetadata}; use borsh::BorshDeserialize; @@ -20,8 +22,7 @@ impl BlockMetaEvent { program_received_time_us: i64, ) -> Self { let metadata = EventMetadata::new( - format!("block_{}_{}", slot, block_hash), - "".to_string(), + Cow::Borrowed(""), slot, block_time_ms / 1000, block_time_ms, diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 0c08216..e16fc8f 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -118,8 +118,6 @@ impl BonkEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = bonk_pool_create_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(metadata.signature.to_string()); Some(Box::new(BonkPoolCreateEvent { metadata, ..event })) } else { None @@ -132,8 +130,6 @@ impl BonkEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = bonk_trade_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, event.pool_state)); if metadata.event_type == EventType::BonkBuyExactIn || metadata.event_type == EventType::BonkBuyExactOut { @@ -166,9 +162,6 @@ impl BonkEventParser { let minimum_amount_out = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - Some(Box::new(BonkTradeEvent { metadata, amount_in, @@ -207,9 +200,6 @@ impl BonkEventParser { let maximum_amount_in = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - Some(Box::new(BonkTradeEvent { metadata, amount_out, @@ -248,9 +238,6 @@ impl BonkEventParser { let minimum_amount_out = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - Some(Box::new(BonkTradeEvent { metadata, amount_in, @@ -289,9 +276,6 @@ impl BonkEventParser { let maximum_amount_in = read_u64_le(data, 8)?; let share_fee_rate = read_u64_le(data, 16)?; - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - Some(Box::new(BonkTradeEvent { metadata, amount_out, @@ -332,9 +316,6 @@ impl BonkEventParser { let curve_param = Self::parse_curve_params(data, &mut offset)?; let vesting_param = Self::parse_vesting_params(data, &mut offset)?; - let mut metadata = metadata; - metadata.set_id(metadata.signature.to_string()); - Some(Box::new(BonkPoolCreateEvent { metadata, payer: accounts[0], @@ -369,9 +350,6 @@ impl BonkEventParser { let vesting_param = Self::parse_vesting_params(data, &mut offset)?; let amm_fee_on = data[offset]; - let mut metadata = metadata; - metadata.set_id(metadata.signature.to_string()); - Some(Box::new(BonkPoolCreateEvent { metadata, payer: accounts[0], @@ -514,9 +492,6 @@ impl BonkEventParser { let quote_lot_size = u64::from_le_bytes(data[8..16].try_into().unwrap()); let market_vault_signer_nonce = data[16]; - let mut metadata = metadata; - metadata.set_id(metadata.signature.to_string()); - Some(Box::new(BonkMigrateToAmmEvent { metadata, base_lot_size, @@ -564,9 +539,6 @@ impl BonkEventParser { accounts: &[Pubkey], metadata: EventMetadata, ) -> Option> { - let mut metadata = metadata; - metadata.set_id(metadata.signature.to_string()); - Some(Box::new(BonkMigrateToCpswapEvent { metadata, payer: accounts[0], diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index 5596f38..0de1634 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -81,8 +81,6 @@ impl PumpFunEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pumpfun_migrate_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, event.user, event.mint)); Some(Box::new(PumpFunMigrateEvent { metadata, ..event })) } else { None @@ -95,11 +93,6 @@ impl PumpFunEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pumpfun_create_token_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.name, event.symbol, event.mint - )); Some(Box::new(PumpFunCreateTokenEvent { metadata, ..event })) } else { None @@ -112,11 +105,6 @@ impl PumpFunEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pumpfun_trade_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.mint, event.user, event.is_buy - )); Some(Box::new(PumpFunTradeEvent { metadata, ..event })) } else { None @@ -151,9 +139,6 @@ impl PumpFunEventParser { Pubkey::default() }; - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}-{}", metadata.signature, name, symbol, accounts[0])); - Some(Box::new(PumpFunCreateTokenEvent { metadata, name: name.to_string(), @@ -180,8 +165,6 @@ impl PumpFunEventParser { } let amount = u64::from_le_bytes(data[0..8].try_into().unwrap()); let max_sol_cost = u64::from_le_bytes(data[8..16].try_into().unwrap()); - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}-{}", metadata.signature, accounts[2], accounts[6], true)); Some(Box::new(PumpFunTradeEvent { metadata, global: accounts[0], @@ -216,9 +199,6 @@ impl PumpFunEventParser { } let amount = u64::from_le_bytes(data[0..8].try_into().unwrap()); let min_sol_output = u64::from_le_bytes(data[8..16].try_into().unwrap()); - let mut metadata = metadata; - metadata - .set_id(format!("{}-{}-{}-{}", metadata.signature, accounts[2], accounts[6], false)); Some(Box::new(PumpFunTradeEvent { metadata, global: accounts[0], @@ -251,8 +231,6 @@ impl PumpFunEventParser { if accounts.len() < 24 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[5], accounts[2])); Some(Box::new(PumpFunMigrateEvent { metadata, global: accounts[0], diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index dc22336..d50372e 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -91,11 +91,6 @@ impl PumpSwapEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pump_swap_buy_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.user, event.pool, event.base_amount_out - )); Some(Box::new(PumpSwapBuyEvent { metadata, ..event })) } else { None @@ -108,11 +103,6 @@ impl PumpSwapEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pump_swap_sell_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.user, event.pool, event.base_amount_in - )); Some(Box::new(PumpSwapSellEvent { metadata, ..event })) } else { None @@ -125,11 +115,6 @@ impl PumpSwapEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pump_swap_create_pool_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.pool, event.creator, event.base_amount_in - )); Some(Box::new(PumpSwapCreatePoolEvent { metadata, ..event })) } else { None @@ -142,11 +127,6 @@ impl PumpSwapEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pump_swap_deposit_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.pool, event.user, event.lp_token_amount_out - )); Some(Box::new(PumpSwapDepositEvent { metadata, ..event })) } else { None @@ -159,11 +139,6 @@ impl PumpSwapEventParser { metadata: EventMetadata, ) -> Option> { if let Some(event) = pump_swap_withdraw_event_log_decode(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.pool, event.user, event.lp_token_amount_in - )); Some(Box::new(PumpSwapWithdrawEvent { metadata, ..event })) } else { None @@ -183,12 +158,6 @@ impl PumpSwapEventParser { let base_amount_out = read_u64_le(data, 0)?; let max_quote_amount_in = read_u64_le(data, 8)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[1], accounts[0], base_amount_out - )); - Some(Box::new(PumpSwapBuyEvent { metadata, base_amount_out, @@ -224,12 +193,6 @@ impl PumpSwapEventParser { let base_amount_in = read_u64_le(data, 0)?; let min_quote_amount_out = read_u64_le(data, 8)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[1], accounts[0], base_amount_in - )); - Some(Box::new(PumpSwapSellEvent { metadata, base_amount_in, @@ -271,12 +234,6 @@ impl PumpSwapEventParser { Pubkey::default() }; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[0], accounts[2], base_amount_in - )); - Some(Box::new(PumpSwapCreatePoolEvent { metadata, index, @@ -311,12 +268,6 @@ impl PumpSwapEventParser { let max_base_amount_in = u64::from_le_bytes(data[8..16].try_into().ok()?); let max_quote_amount_in = u64::from_le_bytes(data[16..24].try_into().ok()?); - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[0], accounts[2], lp_token_amount_out - )); - Some(Box::new(PumpSwapDepositEvent { metadata, lp_token_amount_out, @@ -349,12 +300,6 @@ impl PumpSwapEventParser { let min_base_amount_out = u64::from_le_bytes(data[8..16].try_into().ok()?); let min_quote_amount_out = u64::from_le_bytes(data[16..24].try_into().ok()?); - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[0], accounts[2], lp_token_amount_in - )); - Some(Box::new(PumpSwapWithdrawEvent { metadata, lp_token_amount_in, diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs index 6b8a7ba..f9124e3 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs @@ -102,12 +102,6 @@ impl RaydiumAmmV4EventParser { return None; } - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - Some(Box::new(RaydiumAmmV4WithdrawPnlEvent { metadata, token_program: accounts[0], @@ -141,12 +135,6 @@ impl RaydiumAmmV4EventParser { } let amount = read_u64_le(data, 0)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - Some(Box::new(RaydiumAmmV4WithdrawEvent { metadata, amount, @@ -190,12 +178,6 @@ impl RaydiumAmmV4EventParser { let init_pc_amount = read_u64_le(data, 9)?; let init_coin_amount = read_u64_le(data, 17)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - Some(Box::new(RaydiumAmmV4Initialize2Event { metadata, nonce, @@ -240,12 +222,6 @@ impl RaydiumAmmV4EventParser { let max_pc_amount = read_u64_le(data, 8)?; let base_side = read_u64_le(data, 16)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - Some(Box::new(RaydiumAmmV4DepositEvent { metadata, max_coin_amount, @@ -281,12 +257,6 @@ impl RaydiumAmmV4EventParser { let max_amount_in = read_u64_le(data, 0)?; let amount_out = read_u64_le(data, 8)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - let mut accounts = accounts.to_vec(); if accounts.len() == 17 { // 添加一个默认的 Pubkey 作为 amm_target_orders 的占位符 @@ -334,12 +304,6 @@ impl RaydiumAmmV4EventParser { let amount_in = read_u64_le(data, 0)?; let minimum_amount_out = read_u64_le(data, 8)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - let mut accounts = accounts.to_vec(); if accounts.len() == 17 { // 添加一个默认的 Pubkey 作为 amm_target_orders 的占位符 diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index 7f56274..ced7677 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -124,8 +124,6 @@ impl RaydiumClmmEventParser { if data.len() < 51 || accounts.len() < 22 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumClmmOpenPositionV2Event { metadata, tick_lower_index: read_i32_le(data, 0)?, @@ -172,8 +170,6 @@ impl RaydiumClmmEventParser { if data.len() < 51 || accounts.len() < 20 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumClmmOpenPositionWithToken22NftEvent { metadata, tick_lower_index: read_i32_le(data, 0)?, @@ -217,8 +213,6 @@ impl RaydiumClmmEventParser { if data.len() < 34 || accounts.len() < 15 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumClmmIncreaseLiquidityV2Event { metadata, liquidity: read_u128_le(data, 0)?, @@ -252,8 +246,6 @@ impl RaydiumClmmEventParser { if data.len() < 24 || accounts.len() < 13 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumClmmCreatePoolEvent { metadata, sqrt_price_x64: read_u128_le(data, 0)?, @@ -283,8 +275,6 @@ impl RaydiumClmmEventParser { if data.len() < 32 || accounts.len() < 16 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumClmmDecreaseLiquidityV2Event { metadata, liquidity: read_u128_le(data, 0)?, @@ -319,8 +309,6 @@ impl RaydiumClmmEventParser { if accounts.len() < 6 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumClmmClosePositionEvent { metadata, nft_owner: accounts[0], @@ -347,12 +335,6 @@ impl RaydiumClmmEventParser { let sqrt_price_limit_x64 = read_u128_le(data, 16)?; let is_base_input = read_u8_le(data, 32)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[2], accounts[3], accounts[4] - )); - Some(Box::new(RaydiumClmmSwapEvent { metadata, amount, @@ -387,12 +369,6 @@ impl RaydiumClmmEventParser { let sqrt_price_limit_x64 = read_u128_le(data, 16)?; let is_base_input = read_u8_le(data, 32)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[2], accounts[3], accounts[4] - )); - Some(Box::new(RaydiumClmmSwapV2Event { metadata, amount, diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index f7d2e2e..d89f779 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -92,8 +92,6 @@ impl RaydiumCpmmEventParser { if data.len() < 24 || accounts.len() < 14 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumCpmmWithdrawEvent { metadata, lp_token_amount: read_u64_le(data, 0)?, @@ -125,8 +123,6 @@ impl RaydiumCpmmEventParser { if data.len() < 24 || accounts.len() < 20 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumCpmmInitializeEvent { metadata, init_amount0: read_u64_le(data, 0)?, @@ -164,8 +160,6 @@ impl RaydiumCpmmEventParser { if data.len() < 24 || accounts.len() < 13 { return None; } - let mut metadata = metadata; - metadata.set_id(format!("{}-{}-{}", metadata.signature, accounts[0], accounts[1])); Some(Box::new(RaydiumCpmmDepositEvent { metadata, lp_token_amount: read_u64_le(data, 0)?, @@ -200,12 +194,6 @@ impl RaydiumCpmmEventParser { let amount_in = read_u64_le(data, 0)?; let minimum_amount_out = read_u64_le(data, 8)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - Some(Box::new(RaydiumCpmmSwapEvent { metadata, amount_in, @@ -239,12 +227,6 @@ impl RaydiumCpmmEventParser { let max_amount_in = read_u64_le(data, 0)?; let amount_out = read_u64_le(data, 8)?; - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - Some(Box::new(RaydiumCpmmSwapEvent { metadata, max_amount_in, diff --git a/src/streaming/grpc/subscription.rs b/src/streaming/grpc/subscription.rs index 15579fa..06d4d59 100644 --- a/src/streaming/grpc/subscription.rs +++ b/src/streaming/grpc/subscription.rs @@ -14,7 +14,7 @@ use crate::common::AnyResult; use crate::streaming::common::StreamClientConfig as ClientConfig; use crate::streaming::event_parser::common::filter::EventTypeFilter; -/// 订阅管理器 +/// Subscription manager #[derive(Clone)] pub struct SubscriptionManager { endpoint: String, @@ -23,12 +23,12 @@ pub struct SubscriptionManager { } impl SubscriptionManager { - /// 创建新的订阅管理器 + /// Create a new subscription manager pub fn new(endpoint: String, x_token: Option, config: ClientConfig) -> Self { Self { endpoint, x_token, config } } - /// 创建 gRPC 连接 + /// Create gRPC connection pub async fn connect(&self) -> AnyResult> { let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? .x_token(self.x_token.clone())? @@ -39,20 +39,20 @@ impl SubscriptionManager { Ok(builder.connect().await?) } - /// 创建订阅请求并返回流 + /// Create subscription request and return stream pub async fn subscribe_with_request( &self, transactions: Option, accounts: Option, commitment: Option, - event_type_filter: Option, + event_type_filter: Option<&EventTypeFilter>, ) -> AnyResult<( impl Sink, impl Stream>, SubscribeRequest, )> { let blocks_meta = if event_type_filter.is_some() - && event_type_filter.as_ref().unwrap().include_block_event() + && event_type_filter.unwrap().include_block_event() { hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} } } else if event_type_filter.is_none() { @@ -76,18 +76,18 @@ impl SubscriptionManager { Ok((sink, stream, subscribe_request)) } - /// 创建账户订阅请求并返回流 + /// Create account subscription request and return stream pub fn subscribe_with_account_request( &self, account: Vec, owner: Vec, - event_type_filter: Option, + event_type_filter: Option<&EventTypeFilter>, ) -> Option { if account.len() == 0 && owner.len() == 0 { return None; } if event_type_filter.is_some() - && !event_type_filter.as_ref().unwrap().include_account_event() + && !event_type_filter.unwrap().include_account_event() { return None; } @@ -104,16 +104,16 @@ impl SubscriptionManager { Some(accounts) } - /// 生成订阅请求过滤器 + /// Generate subscription request filter pub fn get_subscribe_request_filter( &self, account_include: Vec, account_exclude: Vec, account_required: Vec, - event_type_filter: Option, + event_type_filter: Option<&EventTypeFilter>, ) -> Option { if event_type_filter.is_some() - && !event_type_filter.as_ref().unwrap().include_transaction_event() + && !event_type_filter.unwrap().include_transaction_event() { return None; } @@ -132,7 +132,7 @@ impl SubscriptionManager { Some(transactions) } - /// 获取配置 + /// Get configuration pub fn get_config(&self) -> &ClientConfig { &self.config } diff --git a/src/streaming/shred/connection.rs b/src/streaming/shred/connection.rs index 9c6bbe3..951993c 100644 --- a/src/streaming/shred/connection.rs +++ b/src/streaming/shred/connection.rs @@ -59,14 +59,6 @@ impl ShredStreamGrpc { Self::new_with_config(endpoint, StreamClientConfig::low_latency()).await } - /// Creates a new ShredStreamClient with asynchronous processing configuration. - /// - /// This is a convenience method that creates a client optimized for high-volume scenarios - /// with balanced throughput and reliability. See `StreamClientConfig::async_processing()` - /// for detailed configuration information. - pub async fn new_async_processing(endpoint: String) -> AnyResult { - Self::new_with_config(endpoint, StreamClientConfig::async_processing()).await - } /// 获取当前配置 pub fn get_config(&self) -> &StreamClientConfig { diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs index 1ec59af..34101c0 100755 --- a/src/streaming/shred_stream.rs +++ b/src/streaming/shred_stream.rs @@ -62,19 +62,16 @@ impl ShredStreamGrpc { msg.slot, chrono::Utc::now().timestamp_micros(), ); - // 异步执行，不阻塞主流，使用带背压控制的方法 - let processor_clone = event_processor_clone.clone(); - tokio::spawn(async move { - if let Err(e) = processor_clone - .process_shred_transaction_with_metrics( - transaction_with_slot, - bot_wallet, - ) - .await - { - error!("Error handling message: {e:?}"); - } - }); + // 直接处理，背压控制在 EventProcessor 内部处理 + if let Err(e) = event_processor_clone + .process_shred_transaction_with_metrics( + transaction_with_slot, + bot_wallet, + ) + .await + { + error!("Error handling message: {e:?}"); + } } } } diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index 7ffb886..f083b8b 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -106,15 +106,6 @@ impl YellowstoneGrpc { Self::new_with_config(endpoint, x_token, StreamClientConfig::low_latency()) } - /// Creates a new YellowstoneGrpcClient with asynchronous processing configuration. - /// - /// This is a convenience method that creates a client optimized for high-volume scenarios - /// with balanced throughput and reliability. See `StreamClientConfig::async_processing()` - /// for detailed configuration information. - pub fn new_async_processing(endpoint: String, x_token: Option) -> AnyResult { - let config = StreamClientConfig::async_processing(); - Self::new_with_config(endpoint, x_token, config) - } /// 获取配置 pub fn get_config(&self) -> &StreamClientConfig { @@ -196,18 +187,18 @@ impl YellowstoneGrpc { transaction_filter.account_include, transaction_filter.account_exclude, transaction_filter.account_required, - event_type_filter.clone(), + event_type_filter.as_ref(), ); let accounts = self.subscription_manager.subscribe_with_account_request( account_filter.account, account_filter.owner, - event_type_filter.clone(), + event_type_filter.as_ref(), ); // 订阅事件 let (mut subscribe_tx, mut stream, subscribe_request) = self .subscription_manager - .subscribe_with_request(transactions, accounts, commitment, event_type_filter.clone()) + .subscribe_with_request(transactions, accounts, commitment, event_type_filter.as_ref()) .await?; // 用 Arc> 包装 subscribe_tx 以支持多线程共享 @@ -224,84 +215,78 @@ impl YellowstoneGrpc { self.config.backpressure.clone(), Some(Arc::new(callback)), ); - let event_processor = Arc::new(event_processor); let stream_handle = tokio::spawn(async move { loop { tokio::select! { message = stream.next() => { match message { Some(Ok(msg)) => { - // 不阻塞地处理消息，使用 tokio::spawn 实现并发 - let event_processor_ref = Arc::clone(&event_processor); - let subscribe_tx_ref = Arc::clone(&subscribe_tx); - tokio::spawn(async move { - let created_at = msg.created_at; - match msg.update_oneof { - Some(UpdateOneof::Account(account)) => { - let account_pretty = AccountPretty::from(account); - log::debug!("Received account: {:?}", account_pretty); - if let Err(e) = event_processor_ref - .process_grpc_event_transaction_with_metrics( - EventPretty::Account(account_pretty), - bot_wallet, - ) - .await - { - error!("Error processing account event: {e:?}"); - } - } - Some(UpdateOneof::BlockMeta(sut)) => { - let block_meta_pretty = - BlockMetaPretty::from((sut, created_at)); - log::debug!("Received block meta: {:?}", block_meta_pretty); - if let Err(e) = event_processor_ref - .process_grpc_event_transaction_with_metrics( - EventPretty::BlockMeta(block_meta_pretty), - bot_wallet, - ) - .await - { - error!("Error processing block meta event: {e:?}"); - } - } - Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = - TransactionPretty::from((sut, created_at)); - log::debug!( - "Received transaction: {} at slot {}", - transaction_pretty.signature, - transaction_pretty.slot - ); - if let Err(e) = event_processor_ref - .process_grpc_event_transaction_with_metrics( - EventPretty::Transaction(transaction_pretty), - bot_wallet, - ) - .await - { - error!("Error processing transaction event: {e:?}"); - } - } - Some(UpdateOneof::Ping(_)) => { - // 只在需要时获取锁，并立即释放 - if let Ok(mut tx_guard) = subscribe_tx_ref.try_lock() { - let _ = tx_guard - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await; - } - log::debug!("service is ping: {}", Local::now()); - } - Some(UpdateOneof::Pong(_)) => { - log::debug!("service is pong: {}", Local::now()); - } - _ => { - log::debug!("Received other message type"); + let created_at = msg.created_at; + match msg.update_oneof { + Some(UpdateOneof::Account(account)) => { + let account_pretty = AccountPretty::from(account); + log::debug!("Received account: {:?}", account_pretty); + if let Err(e) = event_processor + .process_grpc_event_transaction_with_metrics( + EventPretty::Account(account_pretty), + bot_wallet, + ) + .await + { + error!("Error processing account event: {e:?}"); } } - }); + Some(UpdateOneof::BlockMeta(sut)) => { + let block_meta_pretty = + BlockMetaPretty::from((sut, created_at)); + log::debug!("Received block meta: {:?}", block_meta_pretty); + if let Err(e) = event_processor + .process_grpc_event_transaction_with_metrics( + EventPretty::BlockMeta(block_meta_pretty), + bot_wallet, + ) + .await + { + error!("Error processing block meta event: {e:?}"); + } + } + Some(UpdateOneof::Transaction(sut)) => { + let transaction_pretty = + TransactionPretty::from((sut, created_at)); + log::debug!( + "Received transaction: {} at slot {}", + transaction_pretty.signature, + transaction_pretty.slot + ); + if let Err(e) = event_processor + .process_grpc_event_transaction_with_metrics( + EventPretty::Transaction(transaction_pretty), + bot_wallet, + ) + .await + { + error!("Error processing transaction event: {e:?}"); + } + } + Some(UpdateOneof::Ping(_)) => { + // 只在需要时获取锁，并立即释放 + if let Ok(mut tx_guard) = subscribe_tx.try_lock() { + let _ = tx_guard + .send(SubscribeRequest { + ping: Some(SubscribeRequestPing { id: 1 }), + ..Default::default() + }) + .await; + } + log::debug!("service is ping: {}", Local::now()); + } + Some(UpdateOneof::Pong(_)) => { + log::debug!("service is pong: {}", Local::now()); + } + _ => { + log::debug!("Received other message type"); + } + } } Some(Err(error)) => { error!("Stream error: {error:?}"); diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index d7cabcf..030397d 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -64,17 +64,14 @@ impl YellowstoneGrpc { Some(UpdateOneof::Transaction(sut)) => { let transaction_pretty = TransactionPretty::from((sut, created_at)); let event_pretty = EventPretty::Transaction(transaction_pretty); - let callback_clone = callback.clone(); - tokio::spawn(async move { - if let Err(e) = Self::process_system_transaction( - event_pretty, - &*callback_clone, - ) - .await - { - error!("Error processing transaction: {e:?}"); - } - }); + if let Err(e) = Self::process_system_transaction( + event_pretty, + &*callback, + ) + .await + { + error!("Error processing transaction: {e:?}"); + } } Some(UpdateOneof::Ping(_)) => { let _ = subscribe_tx