From 9c311f63c17bdd3af20d75543b2594d81ca2314d Mon Sep 17 00:00:00 2001 From: ysq Date: Sat, 13 Sep 2025 17:10:58 +0800 Subject: [PATCH] refactor: migrate from slot-based to signature-based global state Replace slot-based storage with signature-based storage in GlobalState for improved transaction-level precision and developer address tracking. - Update GlobalState structure and APIs - Modify event processors to use transaction signatures - Maintain lock-free performance characteristics --- Cargo.toml | 2 +- README.md | 4 +- README_CN.md | 4 +- .../event_parser/core/global_state.rs | 168 +++++++++--------- src/streaming/event_parser/core/traits.rs | 15 +- 5 files changed, 97 insertions(+), 96 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index fc328fe..12b0ec0 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.4.7" +version = "0.4.8" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index 6f384c9..b890156 100755 --- a/README.md +++ b/README.md @@ -107,14 +107,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.7" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.8" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.4.7" +solana-streamer-sdk = "0.4.8" ``` ## ⚙️ Configuration System diff --git a/README_CN.md b/README_CN.md index 63457fc..c1cf59a 100644 --- a/README_CN.md +++ b/README_CN.md @@ -107,14 +107,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.7" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.8" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.4.7" +solana-streamer-sdk = "0.4.8" ``` ## ⚙️ 配置系统 diff --git a/src/streaming/event_parser/core/global_state.rs b/src/streaming/event_parser/core/global_state.rs index c14142e..de6dc45 100644 --- a/src/streaming/event_parser/core/global_state.rs +++ b/src/streaming/event_parser/core/global_state.rs @@ -1,26 +1,27 @@ use solana_sdk::pubkey::Pubkey; +use solana_sdk::signature::Signature; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use dashmap::DashMap; use std::collections::BTreeSet; -const MAX_SLOTS: usize = 1000; +const MAX_SIGNATURES: usize = 1000; const CLEANUP_BATCH_SIZE: usize = 100; -/// Slot-based trader addresses, completely lock-free +/// Signature-based trader addresses, completely lock-free #[derive(Default)] -struct SlotAddresses { - /// Developer addresses for this slot +struct SignatureAddresses { + /// Developer addresses for this signature dev_addresses: BTreeSet, - /// Bonk developer addresses for this slot + /// Bonk developer addresses for this signature bonk_dev_addresses: BTreeSet, } -/// High-performance global state with lock-free slot-based storage +/// High-performance global state with lock-free signature-based storage pub struct GlobalState { - /// Slot -> trader addresses mapping (lock-free concurrent hashmap) - slot_data: DashMap, - /// Current slot count for capacity management - slot_count: AtomicUsize, + /// Signature -> trader addresses mapping (lock-free concurrent hashmap) + signature_data: DashMap, + /// Current signature count for capacity management + signature_count: AtomicUsize, /// Generation counter to handle cleanup races generation: AtomicU64, } @@ -29,16 +30,16 @@ impl GlobalState { /// Create a new high-performance global state instance pub fn new() -> Self { Self { - slot_data: DashMap::new(), - slot_count: AtomicUsize::new(0), + signature_data: DashMap::new(), + signature_count: AtomicUsize::new(0), generation: AtomicU64::new(0), } } - /// Lock-free capacity management - cleanup old slots when limit exceeded + /// Lock-free capacity management - cleanup old signatures when limit exceeded fn maybe_cleanup(&self) { - let current_count = self.slot_count.load(Ordering::Relaxed); - if current_count <= MAX_SLOTS { + let current_count = self.signature_count.load(Ordering::Relaxed); + if current_count <= MAX_SIGNATURES { return; } @@ -48,85 +49,84 @@ impl GlobalState { return; // Another thread is cleaning up } - // Collect oldest slots (BTreeMap naturally orders by key) - let mut slots_to_remove: Vec = self.slot_data.iter() + // Collect signatures to remove (random selection for simplicity) + let mut signatures_to_remove: Vec = self.signature_data.iter() .map(|entry| *entry.key()) .collect(); - if slots_to_remove.len() <= MAX_SLOTS { + if signatures_to_remove.len() <= MAX_SIGNATURES { return; // Race condition, already cleaned up } - slots_to_remove.sort_unstable(); - slots_to_remove.truncate(CLEANUP_BATCH_SIZE); + signatures_to_remove.truncate(CLEANUP_BATCH_SIZE); - // Remove old slots atomically - for slot in slots_to_remove { - self.slot_data.remove(&slot); - self.slot_count.fetch_sub(1, Ordering::Relaxed); + // Remove old signatures atomically + for signature in signatures_to_remove { + self.signature_data.remove(&signature); + self.signature_count.fetch_sub(1, Ordering::Relaxed); } } - /// Add developer address for a specific slot (lock-free) - pub fn add_dev_address(&self, slot: u64, address: Pubkey) { + /// Add developer address for a specific signature (lock-free) + pub fn add_dev_address(&self, signature: &Signature, address: Pubkey) { self.maybe_cleanup(); - self.slot_data.entry(slot) + self.signature_data.entry(*signature) .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 + self.signature_count.fetch_add(1, Ordering::Relaxed); + let mut sig_addr = SignatureAddresses::default(); + sig_addr.dev_addresses.insert(address); + sig_addr }); } - /// Add Bonk developer address for a specific slot (lock-free) - pub fn add_bonk_dev_address(&self, slot: u64, address: Pubkey) { + /// Add Bonk developer address for a specific signature (lock-free) + pub fn add_bonk_dev_address(&self, signature: &Signature, address: Pubkey) { self.maybe_cleanup(); - self.slot_data.entry(slot) + self.signature_data.entry(*signature) .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 + self.signature_count.fetch_add(1, Ordering::Relaxed); + let mut sig_addr = SignatureAddresses::default(); + sig_addr.bonk_dev_addresses.insert(address); + sig_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) + /// High-performance: Check if address is a developer address in specific signature (O(log m)) + pub fn is_dev_address_in_signature(&self, signature: &Signature, address: &Pubkey) -> bool { + self.signature_data.get(signature) .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) + /// High-performance: Check if address is a Bonk developer address in specific signature (O(log m)) + pub fn is_bonk_dev_address_in_signature(&self, signature: &Signature, address: &Pubkey) -> bool { + self.signature_data.get(signature) .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) + /// Check if address is a developer address in any signature (lock-free scan, slower) pub fn is_dev_address(&self, address: &Pubkey) -> bool { - self.slot_data.iter().any(|entry| entry.dev_addresses.contains(address)) + self.signature_data.iter().any(|entry| entry.dev_addresses.contains(address)) } - /// Check if address is a Bonk developer address in any slot (lock-free scan, slower) + /// Check if address is a Bonk developer address in any signature (lock-free scan, slower) pub fn is_bonk_dev_address(&self, address: &Pubkey) -> bool { - self.slot_data.iter().any(|entry| entry.bonk_dev_addresses.contains(address)) + self.signature_data.iter().any(|entry| entry.bonk_dev_addresses.contains(address)) } - /// Get all developer addresses from all slots (lock-free aggregation) + /// Get all developer addresses from all signatures (lock-free aggregation) pub fn get_dev_addresses(&self) -> Vec { let mut all_addresses = BTreeSet::new(); - for entry in self.slot_data.iter() { + for entry in self.signature_data.iter() { for addr in &entry.dev_addresses { all_addresses.insert(*addr); } @@ -134,10 +134,10 @@ impl GlobalState { all_addresses.into_iter().collect() } - /// Get all Bonk developer addresses from all slots (lock-free aggregation) + /// Get all Bonk developer addresses from all signatures (lock-free aggregation) pub fn get_bonk_dev_addresses(&self) -> Vec { let mut all_addresses = BTreeSet::new(); - for entry in self.slot_data.iter() { + for entry in self.signature_data.iter() { for addr in &entry.bonk_dev_addresses { all_addresses.insert(*addr); } @@ -145,29 +145,29 @@ impl GlobalState { all_addresses.into_iter().collect() } - /// Get developer addresses for a specific slot - pub fn get_dev_addresses_for_slot(&self, slot: u64) -> Vec { - self.slot_data.get(&slot) + /// Get developer addresses for a specific signature + pub fn get_dev_addresses_for_signature(&self, signature: &Signature) -> Vec { + self.signature_data.get(signature) .map(|entry| entry.dev_addresses.iter().copied().collect()) .unwrap_or_default() } - /// 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) + /// Get Bonk developer addresses for a specific signature + pub fn get_bonk_dev_addresses_for_signature(&self, signature: &Signature) -> Vec { + self.signature_data.get(signature) .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) + /// Get current signature count + pub fn get_signature_count(&self) -> usize { + self.signature_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.signature_data.clear(); + self.signature_count.store(0, Ordering::Relaxed); self.generation.store(0, Ordering::Relaxed); } } @@ -187,9 +187,9 @@ pub fn get_global_state() -> &'static GlobalState { &GLOBAL_STATE } -/// 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: Add developer address for a specific signature +pub fn add_dev_address(signature: &Signature, address: Pubkey) { + get_global_state().add_dev_address(signature, address); } /// Convenience function: Check if address is a developer address @@ -197,9 +197,9 @@ pub fn is_dev_address(address: &Pubkey) -> bool { get_global_state().is_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: Add Bonk developer address for a specific signature +pub fn add_bonk_dev_address(signature: &Signature, address: Pubkey) { + get_global_state().add_bonk_dev_address(signature, address); } /// Convenience function: Check if address is a Bonk developer address @@ -217,27 +217,27 @@ 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 developer addresses for a specific signature +pub fn get_dev_addresses_for_signature(signature: &Signature) -> Vec { + get_global_state().get_dev_addresses_for_signature(signature) } -/// 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 Bonk developer addresses for a specific signature +pub fn get_bonk_dev_addresses_for_signature(signature: &Signature) -> Vec { + get_global_state().get_bonk_dev_addresses_for_signature(signature) } -/// Convenience function: Get current slot count -pub fn get_slot_count() -> usize { - get_global_state().get_slot_count() +/// Convenience function: Get current signature count +pub fn get_signature_count() -> usize { + get_global_state().get_signature_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 developer address in specific signature +pub fn is_dev_address_in_signature(signature: &Signature, address: &Pubkey) -> bool { + get_global_state().is_dev_address_in_signature(signature, 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) +/// High-performance: Check if address is a Bonk developer address in specific signature +pub fn is_bonk_dev_address_in_signature(signature: &Signature, address: &Pubkey) -> bool { + get_global_state().is_bonk_dev_address_in_signature(signature, address) } diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 514fff8..5ef7c95 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -16,7 +16,8 @@ use std::time::Instant; use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo; use super::global_state::{ - add_bonk_dev_address, add_dev_address, is_bonk_dev_address, is_dev_address, + add_bonk_dev_address, add_dev_address, + is_bonk_dev_address_in_signature, is_dev_address_in_signature, }; use crate::streaming::common::simd_utils::SimdUtils; @@ -1389,14 +1390,14 @@ fn process_event( mut event: Box, bot_wallet: Option, ) -> Box { - let slot = event.slot(); + let signature = *event.signature(); // Copy the signature to avoid borrowing issues if let Some(token_info) = event.as_any().downcast_ref::() { - add_dev_address(slot, token_info.user); + add_dev_address(&signature, token_info.user); if token_info.creator != Pubkey::default() && token_info.creator != token_info.user { - add_dev_address(slot, token_info.creator); + add_dev_address(&signature, 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) { + if is_dev_address_in_signature(&signature, &trade_info.user) || is_dev_address_in_signature(&signature, &trade_info.creator) { trade_info.is_dev_create_token_trade = true; } else if Some(trade_info.user) == bot_wallet { trade_info.is_bot = true; @@ -1422,9 +1423,9 @@ fn process_event( trade_info.user_quote_amount_out; } } else if let Some(pool_info) = event.as_any().downcast_ref::() { - add_bonk_dev_address(slot, pool_info.creator); + add_bonk_dev_address(&signature, pool_info.creator); } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { - if is_bonk_dev_address(&trade_info.payer) { + if is_bonk_dev_address_in_signature(&signature, &trade_info.payer) { trade_info.is_dev_create_token_trade = true; } else if Some(trade_info.payer) == bot_wallet { trade_info.is_bot = true;