From 6dab47f8f987b16342527000487b8fdfcab41c30 Mon Sep 17 00:00:00 2001 From: 0xfnzero <0xfnzero@users.noreply.github.com> Date: Tue, 19 May 2026 06:21:37 +0800 Subject: [PATCH] Optimize SWQOS dual fee lane submission --- src/client/mod.rs | 31 ++- src/common/gas_fee_strategy.rs | 189 +++++++++++++++- src/common/sdk_log.rs | 22 +- src/common/types.rs | 10 +- src/instruction/pumpfun.rs | 33 ++- src/instruction/utils/pumpfun.rs | 48 ++-- src/trading/core/async_executor.rs | 351 +++++++++++++++++++++++------ src/trading/core/executor.rs | 33 +-- src/trading/core/traits.rs | 4 +- 9 files changed, 581 insertions(+), 140 deletions(-) diff --git a/src/client/mod.rs b/src/client/mod.rs index efb846f..2d54d0b 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -14,6 +14,7 @@ use crate::constants::WSOL_TOKEN_ACCOUNT; use crate::swqos::common::TradeError; use crate::swqos::SwqosClient; use crate::swqos::SwqosConfig; +use crate::swqos::SwqosType; use crate::swqos::TradeType; use crate::trading::core::params::BonkParams; use crate::trading::core::params::DexParamEnum; @@ -83,7 +84,7 @@ pub struct TradingInfrastructure { pub swqos_clients: Arc>>, /// Configuration used to create this infrastructure pub config: InfrastructureConfig, - /// Precomputed at init: min(swqos_clients.len(), 2/3 * num_cores). Not computed on trade hot path. + /// Precomputed at init: min(max SWQOS submit lanes, 2/3 * num_cores). Not computed on trade hot path. pub max_sender_concurrency: usize, /// Precomputed at init: first max_sender_concurrency CoreIds for job affinity. Empty if no cores. Not computed on trade hot path. pub effective_core_ids: Arc>, @@ -225,11 +226,21 @@ impl TradingInfrastructure { eprintln!("ℹ️ SWQOS 通道已就绪: {} 条 → [{}]", swqos_clients.len(), labels.join(", ")); } - let swqos_count = swqos_clients.len(); + let max_submit_lanes = swqos_clients + .iter() + .map(|client| { + if matches!(client.get_swqos_type(), SwqosType::Default) { + 1usize + } else { + 2usize + } + }) + .sum::() + .max(1); let (max_sender_concurrency, effective_core_ids) = { let num_cores = core_affinity::get_core_ids().map(|c| c.len()).unwrap_or(0); let max_by_cores = (num_cores * 2 / 3).max(1); - let cap = swqos_count.min(max_by_cores).max(1); + let cap = max_submit_lanes.min(max_by_cores).max(1); let ids = core_affinity::get_core_ids() .map(|all| { let v: Vec<_> = all.into_iter().collect(); @@ -713,7 +724,7 @@ impl TradingClient { /// **Advanced.** Use dedicated OS threads for sender pool (and optionally pin to cores). /// By default the SDK uses a shared tokio pool; this can reduce scheduling contention when sending many txs. - /// Concurrency and core count are capped internally (≤ swqos count, ≤ 2/3 of CPU cores). + /// Concurrency and core count are capped internally (≤ max submit lanes, ≤ 2/3 of CPU cores). /// - `None`: keep default (shared tokio pool). /// - `Some(vec![])`: dedicated threads with default count, no core pinning. /// - `Some(indices)`: dedicated threads pinned to those core indices (trimmed to cap). @@ -875,7 +886,11 @@ impl TradingClient { let swap_result = executor.swap(buy_params).await; let result = swap_result.map(|(success, sigs, err, timings)| { - (success, sigs, err.map(TradeError::from), timings) + let legacy_timings = timings + .into_iter() + .map(|timing| (timing.swqos_type, timing.submit_done_us)) + .collect(); + (success, sigs, err.map(TradeError::from), legacy_timings) }); result } @@ -987,7 +1002,11 @@ impl TradingClient { let swap_result = executor.swap(sell_params).await; let result = swap_result.map(|(success, sigs, err, timings)| { - (success, sigs, err.map(TradeError::from), timings) + let legacy_timings = timings + .into_iter() + .map(|timing| (timing.swqos_type, timing.submit_done_us)) + .collect(); + (success, sigs, err.map(TradeError::from), legacy_timings) }); result } diff --git a/src/common/gas_fee_strategy.rs b/src/common/gas_fee_strategy.rs index a0172a8..19ab6c8 100644 --- a/src/common/gas_fee_strategy.rs +++ b/src/common/gas_fee_strategy.rs @@ -1,6 +1,6 @@ use crate::swqos::{SwqosType, TradeType}; use arc_swap::ArcSwap; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -14,6 +14,15 @@ impl GasFeeStrategyType { pub fn values() -> Vec { vec![Self::Normal, Self::LowTipHighCuPrice, Self::HighTipLowCuPrice] } + + #[inline] + pub fn as_str(&self) -> &'static str { + match self { + Self::Normal => "Normal", + Self::LowTipHighCuPrice => "LowTipHighCuPrice", + Self::HighTipLowCuPrice => "HighTipLowCuPrice", + } + } } #[derive(Debug, Clone, Copy)] @@ -84,6 +93,33 @@ impl GasFeeStrategy { ); } + /// 设置 Default/RPC 的优先费-only 策略。Default 没有 relay tip account， + /// 但仍应携带 ComputeBudget 优先费。 + pub fn set_default_rpc_fee_strategy( + &self, + buy_cu_limit: u32, + sell_cu_limit: u32, + buy_cu_price: u64, + sell_cu_price: u64, + ) { + self.set( + SwqosType::Default, + TradeType::Buy, + GasFeeStrategyType::Normal, + buy_cu_limit, + buy_cu_price, + 0.0, + ); + self.set( + SwqosType::Default, + TradeType::Sell, + GasFeeStrategyType::Normal, + sell_cu_limit, + sell_cu_price, + 0.0, + ); + } + /// 为多个服务类型添加高低费率策略，会移除(SwqosType,TradeType)的默认策略。 /// Add high-low fee strategies for multiple service types, Will remove the default strategy of (SwqosType,TradeType) pub fn set_high_low_fee_strategies( @@ -271,7 +307,7 @@ impl GasFeeStrategy { ) -> Vec<(SwqosType, GasFeeStrategyType, GasFeeStrategyValue)> { let strategies = self.strategies.load(); let mut result = Vec::new(); - let mut swqos_types = std::collections::HashSet::new(); + let mut swqos_types = HashSet::new(); for (swqos_type, t_type, _) in strategies.keys() { if *t_type == trade_type { swqos_types.insert(*swqos_type); @@ -298,10 +334,17 @@ impl GasFeeStrategy { /// 动态更新买入小费（保持其他参数不变） /// Dynamically update buy tip (keep other parameters unchanged) pub fn update_buy_tip(&self, buy_tip: f64) { + self.update_buy_tip_for_strategy(GasFeeStrategyType::Normal, buy_tip); + } + + /// 动态更新指定买入策略的小费（保持其他参数不变）。 + /// Dynamic updates should generally target Normal only; updating all strategies would + /// collapse the low-tip/high-tip dual-lane spread into the same tip. + pub fn update_buy_tip_for_strategy(&self, strategy_type: GasFeeStrategyType, buy_tip: f64) { self.strategies.rcu(|current_map| { let mut new_map = (**current_map).clone(); - for ((_swqos_type, trade_type, _strategy_type), value) in new_map.iter_mut() { - if *trade_type == TradeType::Buy { + for ((_swqos_type, trade_type, s_type), value) in new_map.iter_mut() { + if *trade_type == TradeType::Buy && *s_type == strategy_type { value.tip = buy_tip; } } @@ -312,10 +355,15 @@ impl GasFeeStrategy { /// 动态更新卖出小费（保持其他参数不变） /// Dynamically update sell tip (keep other parameters unchanged) pub fn update_sell_tip(&self, sell_tip: f64) { + self.update_sell_tip_for_strategy(GasFeeStrategyType::Normal, sell_tip); + } + + /// 动态更新指定卖出策略的小费（保持其他参数不变）。 + pub fn update_sell_tip_for_strategy(&self, strategy_type: GasFeeStrategyType, sell_tip: f64) { self.strategies.rcu(|current_map| { let mut new_map = (**current_map).clone(); - for ((_swqos_type, trade_type, _strategy_type), value) in new_map.iter_mut() { - if *trade_type == TradeType::Sell { + for ((_swqos_type, trade_type, s_type), value) in new_map.iter_mut() { + if *trade_type == TradeType::Sell && *s_type == strategy_type { value.tip = sell_tip; } } @@ -326,10 +374,19 @@ impl GasFeeStrategy { /// 动态更新买入优先费（保持其他参数不变） /// Dynamically update buy compute unit price (keep other parameters unchanged) pub fn update_buy_cu_price(&self, buy_cu_price: u64) { + self.update_buy_cu_price_for_strategy(GasFeeStrategyType::Normal, buy_cu_price); + } + + /// 动态更新指定买入策略的优先费（保持其他参数不变）。 + pub fn update_buy_cu_price_for_strategy( + &self, + strategy_type: GasFeeStrategyType, + buy_cu_price: u64, + ) { self.strategies.rcu(|current_map| { let mut new_map = (**current_map).clone(); - for ((_swqos_type, trade_type, _strategy_type), value) in new_map.iter_mut() { - if *trade_type == TradeType::Buy { + for ((_swqos_type, trade_type, s_type), value) in new_map.iter_mut() { + if *trade_type == TradeType::Buy && *s_type == strategy_type { value.cu_price = buy_cu_price; } } @@ -340,10 +397,19 @@ impl GasFeeStrategy { /// 动态更新卖出优先费（保持其他参数不变） /// Dynamically update sell compute unit price (keep other parameters unchanged) pub fn update_sell_cu_price(&self, sell_cu_price: u64) { + self.update_sell_cu_price_for_strategy(GasFeeStrategyType::Normal, sell_cu_price); + } + + /// 动态更新指定卖出策略的优先费（保持其他参数不变）。 + pub fn update_sell_cu_price_for_strategy( + &self, + strategy_type: GasFeeStrategyType, + sell_cu_price: u64, + ) { self.strategies.rcu(|current_map| { let mut new_map = (**current_map).clone(); - for ((_swqos_type, trade_type, _strategy_type), value) in new_map.iter_mut() { - if *trade_type == TradeType::Sell { + for ((_swqos_type, trade_type, s_type), value) in new_map.iter_mut() { + if *trade_type == TradeType::Sell && *s_type == strategy_type { value.cu_price = sell_cu_price; } } @@ -365,3 +431,106 @@ impl GasFeeStrategy { } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn find_strategy( + strategies: &[(SwqosType, GasFeeStrategyType, GasFeeStrategyValue)], + swqos_type: SwqosType, + strategy_type: GasFeeStrategyType, + ) -> GasFeeStrategyValue { + strategies + .iter() + .find(|(s, t, _)| *s == swqos_type && *t == strategy_type) + .map(|(_, _, v)| *v) + .expect("strategy exists") + } + + #[test] + fn high_low_fee_strategy_expands_two_lanes_per_swqos() { + let strategy = GasFeeStrategy::new(); + + strategy.set_high_low_fee_strategies( + &[SwqosType::Jito, SwqosType::Helius], + TradeType::Buy, + 100_000, + 180_000, + 400_000, + 0.002, + 0.005, + ); + + let strategies = strategy.get_strategies(TradeType::Buy); + assert_eq!(strategies.len(), 4); + + for swqos_type in [SwqosType::Jito, SwqosType::Helius] { + let low_tip_high_cu = + find_strategy(&strategies, swqos_type, GasFeeStrategyType::LowTipHighCuPrice); + assert_eq!(low_tip_high_cu.cu_limit, 100_000); + assert_eq!(low_tip_high_cu.cu_price, 400_000); + assert_eq!(low_tip_high_cu.tip, 0.002); + + let high_tip_low_cu = + find_strategy(&strategies, swqos_type, GasFeeStrategyType::HighTipLowCuPrice); + assert_eq!(high_tip_low_cu.cu_limit, 100_000); + assert_eq!(high_tip_low_cu.cu_price, 180_000); + assert_eq!(high_tip_low_cu.tip, 0.005); + } + } + + #[test] + fn dynamic_updates_do_not_collapse_dual_lane_fees() { + let strategy = GasFeeStrategy::new(); + + strategy.set_high_low_fee_strategy( + SwqosType::Jito, + TradeType::Buy, + 100_000, + 180_000, + 400_000, + 0.002, + 0.005, + ); + + strategy.update_buy_tip(0.009); + strategy.update_buy_cu_price(999_999); + + let strategies = strategy.get_strategies(TradeType::Buy); + let low_tip_high_cu = + find_strategy(&strategies, SwqosType::Jito, GasFeeStrategyType::LowTipHighCuPrice); + let high_tip_low_cu = + find_strategy(&strategies, SwqosType::Jito, GasFeeStrategyType::HighTipLowCuPrice); + + assert_eq!(low_tip_high_cu.cu_price, 400_000); + assert_eq!(low_tip_high_cu.tip, 0.002); + assert_eq!(high_tip_low_cu.cu_price, 180_000); + assert_eq!(high_tip_low_cu.tip, 0.005); + } + + #[test] + fn default_rpc_strategy_uses_priority_fee_without_tip() { + let strategy = GasFeeStrategy::new(); + + strategy.set_default_rpc_fee_strategy(100_000, 90_000, 700_000, 800_000); + + let buy = find_strategy( + &strategy.get_strategies(TradeType::Buy), + SwqosType::Default, + GasFeeStrategyType::Normal, + ); + assert_eq!(buy.cu_limit, 100_000); + assert_eq!(buy.cu_price, 700_000); + assert_eq!(buy.tip, 0.0); + + let sell = find_strategy( + &strategy.get_strategies(TradeType::Sell), + SwqosType::Default, + GasFeeStrategyType::Normal, + ); + assert_eq!(sell.cu_limit, 90_000); + assert_eq!(sell.cu_price, 800_000); + assert_eq!(sell.tip, 0.0); + } +} diff --git a/src/common/sdk_log.rs b/src/common/sdk_log.rs index 8fc8d18..ca6379b 100644 --- a/src/common/sdk_log.rs +++ b/src/common/sdk_log.rs @@ -84,7 +84,7 @@ pub fn print_sdk_timing_block( start_us: Option, build_end_us: Option, before_submit_us: Option, - submit_timings: &[(crate::swqos::SwqosType, i64)], + submit_timings: &[crate::common::SwqosSubmitTiming], confirm_us: Option, ) { println!(); @@ -112,13 +112,14 @@ pub fn print_sdk_timing_block( } if let Some(confirm_done_us) = confirm_us { let total_ms = (confirm_done_us - start_us) as f64 / 1000.0; - for (swqos_type, submit_done_us) in submit_timings { - let submit_ms = (*submit_done_us - start_us).max(0) as f64 / 1000.0; - let confirmed_ms = (confirm_done_us - *submit_done_us).max(0) as f64 / 1000.0; + for timing in submit_timings { + let submit_ms = (timing.submit_done_us - start_us).max(0) as f64 / 1000.0; + let confirmed_ms = (confirm_done_us - timing.submit_done_us).max(0) as f64 / 1000.0; println!( - " [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", - swqos_type.as_str(), + " [SDK][{:width$}] {} {} submit_done: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", + timing.swqos_type.as_str(), dir, + timing.strategy_type.as_str(), submit_ms, confirmed_ms, total_ms, @@ -126,12 +127,13 @@ pub fn print_sdk_timing_block( ); } } else { - for (swqos_type, submit_done_us) in submit_timings { - let submit_ms = (*submit_done_us - start_us).max(0) as f64 / 1000.0; + for timing in submit_timings { + let submit_ms = (timing.submit_done_us - start_us).max(0) as f64 / 1000.0; println!( - " [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: -, total: {:.4} ms", - swqos_type.as_str(), + " [SDK][{:width$}] {} {} submit_done: {:.4} ms, confirmed: -, total: {:.4} ms", + timing.swqos_type.as_str(), dir, + timing.strategy_type.as_str(), submit_ms, submit_ms, width = SWQOS_LABEL_WIDTH diff --git a/src/common/types.rs b/src/common/types.rs index 77835e9..759d1cb 100755 --- a/src/common/types.rs +++ b/src/common/types.rs @@ -1,4 +1,5 @@ -use crate::swqos::SwqosConfig; +use crate::common::GasFeeStrategyType; +use crate::swqos::{SwqosConfig, SwqosType}; use solana_commitment_config::CommitmentConfig; use std::hash::{Hash, Hasher}; @@ -75,6 +76,13 @@ impl PartialEq for InfrastructureConfig { impl Eq for InfrastructureConfig {} +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SwqosSubmitTiming { + pub swqos_type: SwqosType, + pub strategy_type: GasFeeStrategyType, + pub submit_done_us: i64, +} + #[derive(Debug, Clone)] pub struct TradeConfig { pub rpc_url: String, diff --git a/src/instruction/pumpfun.rs b/src/instruction/pumpfun.rs index 6a85605..7222431 100644 --- a/src/instruction/pumpfun.rs +++ b/src/instruction/pumpfun.rs @@ -34,13 +34,20 @@ use solana_sdk::instruction::AccountMeta; use solana_sdk::{instruction::Instruction, pubkey::Pubkey, signer::Signer}; #[inline] -fn effective_pump_mint_token_program(_mint: &Pubkey, protocol_params: &PumpFunParams) -> Pubkey { +fn is_pump_suffix_mint(mint: &Pubkey) -> bool { + mint.to_string().ends_with("pump") +} + +#[inline] +fn effective_pump_mint_token_program(mint: &Pubkey, protocol_params: &PumpFunParams) -> Pubkey { let tp = protocol_params.token_program; - if tp == Pubkey::default() { - TOKEN_PROGRAM_2022 - } else { - tp + if tp == Pubkey::default() || tp == TOKEN_PROGRAM_2022 { + return TOKEN_PROGRAM_2022; } + if is_pump_suffix_mint(mint) { + return TOKEN_PROGRAM_2022; + } + tp } /// Resolve quote mint and its token program from PumpFunParams. @@ -766,7 +773,7 @@ mod tests { use super::*; use crate::{ common::{bonding_curve::BondingCurveAccount, GasFeeStrategy}, - constants::TOKEN_PROGRAM, + constants::{TOKEN_PROGRAM, TOKEN_PROGRAM_2022}, trading::core::params::{DexParamEnum, PumpFunParams, SwapParams}, }; use solana_sdk::signature::Keypair; @@ -850,17 +857,27 @@ mod tests { } #[test] - fn pump_suffix_buy_respects_explicit_legacy_token_program() { + fn pump_suffix_buy_forces_token_2022_even_when_params_legacy() { crate::common::seed::set_default_rents(); let params = swap_params_for_buy(pump_mint(), TOKEN_PROGRAM); let instructions = build_buy_v1(¶ms).unwrap(); assert_eq!(instructions.len(), 3); - assert_eq!(instructions[2].accounts[8].pubkey, TOKEN_PROGRAM); + assert_eq!(instructions[2].accounts[8].pubkey, TOKEN_PROGRAM_2022); + assert_eq!(instructions[1].program_id, TOKEN_PROGRAM_2022); assert_eq!(instructions[2].accounts.len(), 18); assert_eq!(instructions[2].data.len(), 25); } + #[test] + fn non_pump_buy_respects_explicit_legacy_token_program() { + crate::common::seed::set_default_rents(); + let params = swap_params_for_buy(Pubkey::new_unique(), TOKEN_PROGRAM); + let instructions = build_buy_v1(¶ms).unwrap(); + + assert_eq!(instructions[2].accounts[8].pubkey, TOKEN_PROGRAM); + } + #[test] fn pumpfun_v1_fixed_output_uses_buy_with_max_input_budget() { let mut params = swap_params_for_buy(pump_mint(), TOKEN_PROGRAM); diff --git a/src/instruction/utils/pumpfun.rs b/src/instruction/utils/pumpfun.rs index 6ced69f..3d753e5 100644 --- a/src/instruction/utils/pumpfun.rs +++ b/src/instruction/utils/pumpfun.rs @@ -212,7 +212,7 @@ pub fn is_amm_fee_recipient(pubkey: &Pubkey) -> bool { #[inline] pub fn is_standard_bonding_fee_recipient(pubkey: &Pubkey) -> bool { - *pubkey == global_constants::FEE_RECIPIENT || is_amm_fee_recipient(pubkey) + *pubkey == global_constants::FEE_RECIPIENT } #[inline] @@ -242,11 +242,12 @@ pub fn reconcile_mayhem_mode_for_trade( #[inline] pub fn fee_recipient_ok_for_bonding_curve_mode(pk: &Pubkey, is_mayhem_mode: bool) -> bool { let is_m = is_mayhem_fee_recipient(pk); + let is_amm = is_amm_fee_recipient(pk); let is_s = is_standard_bonding_fee_recipient(pk); if is_mayhem_mode { - !(is_s && !is_m) + is_m || (!is_s && !is_amm && *pk != Pubkey::default()) } else { - !(is_m && !is_s) + is_s || (!is_m && !is_amm && *pk != Pubkey::default()) } } @@ -260,18 +261,10 @@ pub fn get_mayhem_fee_recipient_meta_random() -> AccountMeta { #[inline] pub fn get_standard_fee_recipient_meta_random() -> AccountMeta { - const POOL: &[Pubkey] = &[ - global_constants::FEE_RECIPIENT, - global_constants::PUMPFUN_AMM_FEE_1, - global_constants::PUMPFUN_AMM_FEE_2, - global_constants::PUMPFUN_AMM_FEE_3, - global_constants::PUMPFUN_AMM_FEE_4, - global_constants::PUMPFUN_AMM_FEE_5, - global_constants::PUMPFUN_AMM_FEE_6, - global_constants::PUMPFUN_AMM_FEE_7, - ]; - let recipient = *POOL.choose(&mut rand::rng()).unwrap_or(&global_constants::FEE_RECIPIENT); - AccountMeta { pubkey: recipient, is_signer: false, is_writable: true } + // Historical name kept for API compatibility. Do not randomize across static AMM fee + // recipients for bonding-curve buy/sell; stale AMM protocol fee accounts can fail + // Pump.fun authorization with error 6000 when Global has rotated. + AccountMeta { pubkey: global_constants::FEE_RECIPIENT, is_signer: false, is_writable: true } } #[inline] @@ -707,12 +700,12 @@ mod tests { #[test] fn reconcile_mayhem_prefers_fee_when_log_says_true_but_fee_is_standard_pool() { - let fee = global_constants::PUMPFUN_AMM_FEE_4; + let fee = global_constants::FEE_RECIPIENT; assert!(!reconcile_mayhem_mode_for_trade(Some(true), &fee)); } #[test] - fn pump_fee_meta_rejects_standard_fee_when_building_mayhem_ix() { + fn pump_fee_meta_rejects_amm_fee_when_building_mayhem_ix() { let fee = global_constants::PUMPFUN_AMM_FEE_4; let m = pump_fun_fee_recipient_meta(fee, true); assert!( @@ -723,11 +716,28 @@ mod tests { } #[test] - fn pump_fee_meta_uses_observed_standard_fee_for_standard_ix() { - let fee = global_constants::PUMPFUN_AMM_FEE_7; + fn pump_fee_meta_uses_observed_non_amm_fee_for_standard_ix() { + let fee = Pubkey::new_unique(); let m = pump_fun_fee_recipient_meta(fee, false); assert_eq!(m.pubkey, fee); assert!(m.is_writable); assert!(!m.is_signer); } + + #[test] + fn pump_fee_meta_rejects_amm_fee_for_standard_ix() { + let fee = global_constants::PUMPFUN_AMM_FEE_7; + let m = pump_fun_fee_recipient_meta(fee, false); + assert_eq!(m.pubkey, global_constants::FEE_RECIPIENT); + assert!(m.is_writable); + assert!(!m.is_signer); + } + + #[test] + fn pump_fee_meta_default_standard_uses_main_fee_recipient() { + let m = pump_fun_fee_recipient_meta(Pubkey::default(), false); + assert_eq!(m.pubkey, global_constants::FEE_RECIPIENT); + assert!(m.is_writable); + assert!(!m.is_signer); + } } diff --git a/src/trading/core/async_executor.rs b/src/trading/core/async_executor.rs index b64ccf9..9f1ff79 100644 --- a/src/trading/core/async_executor.rs +++ b/src/trading/core/async_executor.rs @@ -35,8 +35,8 @@ use fnv::FnvHasher; type FnvHashMap = HashMap>; use crate::{ - common::nonce_cache::DurableNonceInfo, - common::GasFeeStrategy, + common::gas_fee_strategy::{GasFeeStrategyType, GasFeeStrategyValue}, + common::{nonce_cache::DurableNonceInfo, GasFeeStrategy, SwqosSubmitTiming}, swqos::{SwqosClient, SwqosType, TradeType}, trading::core::params::SenderConcurrencyConfig, trading::{common::build_transaction, MiddlewareManager}, @@ -71,6 +71,7 @@ struct SwqosJob { tip_account: Arc, swqos_client: Arc, swqos_type: SwqosType, + strategy_type: GasFeeStrategyType, core_id: Option, use_affinity: bool, } @@ -107,6 +108,7 @@ async fn run_one_swqos_job(job: SwqosJob) { signature: Signature::default(), error: Some(e), swqos_type: job.swqos_type, + strategy_type: job.strategy_type, landed_on_chain: false, submit_done_us: crate::common::clock::now_micros(), }); @@ -136,6 +138,7 @@ async fn run_one_swqos_job(job: SwqosJob) { signature: sig, error: err, swqos_type: job.swqos_type, + strategy_type: job.strategy_type, landed_on_chain, submit_done_us: crate::common::clock::now_micros(), }); @@ -153,33 +156,31 @@ async fn swqos_worker_loop(queue: Arc>, notify: Arc static SWQOS_QUEUE: OnceCell>> = OnceCell::new(); static SWQOS_NOTIFY: OnceCell> = OnceCell::new(); -static SWQOS_WORKERS_STARTED: AtomicBool = AtomicBool::new(false); +static SWQOS_WORKER_COUNT: AtomicUsize = AtomicUsize::new(0); /// Dedicated OS-thread sender pool. Queue and notify are in OnceCell so hot path never takes a lock after init. static DEDICATED_QUEUE: OnceCell>> = OnceCell::new(); static DEDICATED_NOTIFY: OnceCell> = OnceCell::new(); +static DEDICATED_WORKER_COUNT: AtomicUsize = AtomicUsize::new(0); /// JoinHandles kept so dedicated threads are not detached; only touched during init under lock. static DEDICATED_INIT: Mutex>>> = Mutex::new(None); -fn ensure_dedicated_pool( +fn desired_dedicated_workers( sender_thread_cores: Option<&[usize]>, max_sender_concurrency: usize, -) -> (Arc>, Arc) { - if let (Some(q), Some(n)) = (DEDICATED_QUEUE.get(), DEDICATED_NOTIFY.get()) { - return (q.clone(), n.clone()); - } - let mut guard = DEDICATED_INIT.lock(); - if let (Some(q), Some(n)) = (DEDICATED_QUEUE.get(), DEDICATED_NOTIFY.get()) { - return (q.clone(), n.clone()); - } - let n = sender_thread_cores +) -> usize { + sender_thread_cores .map(|v| v.len().min(max_sender_concurrency)) .unwrap_or_else(|| SWQOS_DEDICATED_DEFAULT_THREADS.min(max_sender_concurrency)) .min(32) - .max(1); - let queue = Arc::new(ArrayQueue::new(SWQOS_QUEUE_CAP)); - let notify = Arc::new(Notify::new()); - let core_ids: Vec = core_affinity::get_core_ids() + .max(1) +} + +fn dedicated_core_ids( + sender_thread_cores: Option<&[usize]>, + n: usize, +) -> Vec { + core_affinity::get_core_ids() .map(|all_ids| { sender_thread_cores .map(|indices| { @@ -187,9 +188,83 @@ fn ensure_dedicated_pool( }) .unwrap_or_else(|| all_ids.into_iter().take(n).collect()) }) - .unwrap_or_default(); - let mut handles = Vec::with_capacity(n); - for i in 0..n { + .unwrap_or_default() +} + +fn ensure_dedicated_pool( + sender_thread_cores: Option<&[usize]>, + max_sender_concurrency: usize, +) -> (Arc>, Arc) { + let target_workers = desired_dedicated_workers(sender_thread_cores, max_sender_concurrency); + if let (Some(q), Some(n)) = (DEDICATED_QUEUE.get(), DEDICATED_NOTIFY.get()) { + if DEDICATED_WORKER_COUNT.load(Ordering::Acquire) >= target_workers { + return (q.clone(), n.clone()); + } + ensure_dedicated_worker_count(q.clone(), n.clone(), sender_thread_cores, target_workers); + return (q.clone(), n.clone()); + } + let mut guard = DEDICATED_INIT.lock(); + if let (Some(q), Some(n)) = (DEDICATED_QUEUE.get(), DEDICATED_NOTIFY.get()) { + if DEDICATED_WORKER_COUNT.load(Ordering::Acquire) < target_workers { + ensure_dedicated_worker_count_locked( + q.clone(), + n.clone(), + sender_thread_cores, + target_workers, + &mut guard, + ); + } + return (q.clone(), n.clone()); + } + let queue = Arc::new(ArrayQueue::new(SWQOS_QUEUE_CAP)); + let notify = Arc::new(Notify::new()); + let _ = DEDICATED_QUEUE.set(queue.clone()); + let _ = DEDICATED_NOTIFY.set(notify.clone()); + *guard = Some(Vec::with_capacity(target_workers)); + ensure_dedicated_worker_count_locked( + queue.clone(), + notify.clone(), + sender_thread_cores, + target_workers, + &mut guard, + ); + (queue, notify) +} + +fn ensure_dedicated_worker_count( + queue: Arc>, + notify: Arc, + sender_thread_cores: Option<&[usize]>, + target_workers: usize, +) { + if DEDICATED_WORKER_COUNT.load(Ordering::Acquire) >= target_workers { + return; + } + let mut guard = DEDICATED_INIT.lock(); + ensure_dedicated_worker_count_locked( + queue, + notify, + sender_thread_cores, + target_workers, + &mut guard, + ); +} + +fn ensure_dedicated_worker_count_locked( + queue: Arc>, + notify: Arc, + sender_thread_cores: Option<&[usize]>, + target_workers: usize, + guard: &mut Option>>, +) { + let current = DEDICATED_WORKER_COUNT.load(Ordering::Acquire); + if current >= target_workers { + return; + } + let core_ids = dedicated_core_ids(sender_thread_cores, target_workers); + let handles = guard.get_or_insert_with(Vec::new); + handles.reserve(target_workers.saturating_sub(current)); + for i in current..target_workers { let queue = queue.clone(); let notify = notify.clone(); let core_id = core_ids.get(i).cloned(); @@ -205,19 +280,23 @@ fn ensure_dedicated_pool( }); handles.push(handle); } - let _ = DEDICATED_QUEUE.set(queue.clone()); - let _ = DEDICATED_NOTIFY.set(notify.clone()); - *guard = Some(handles); - (queue, notify) + DEDICATED_WORKER_COUNT.store(target_workers, Ordering::Release); } fn ensure_swqos_pool(queue: Arc>, max_sender_concurrency: usize) { - if SWQOS_WORKERS_STARTED.swap(true, Ordering::AcqRel) { + let n = SWQOS_POOL_WORKERS.min(max_sender_concurrency).max(1); + let mut current = SWQOS_WORKER_COUNT.load(Ordering::Acquire); + while current < n { + match SWQOS_WORKER_COUNT.compare_exchange(current, n, Ordering::AcqRel, Ordering::Acquire) { + Ok(_) => break, + Err(actual) => current = actual, + } + } + if current >= n { return; } - let n = SWQOS_POOL_WORKERS.min(max_sender_concurrency).max(1); let notify = SWQOS_NOTIFY.get_or_init(|| Arc::new(Notify::new())).clone(); - for _ in 0..n { + for _ in current..n { tokio::spawn(swqos_worker_loop(queue.clone(), notify.clone())); } } @@ -228,6 +307,7 @@ struct TaskResult { signature: Signature, error: Option, swqos_type: SwqosType, + strategy_type: GasFeeStrategyType, landed_on_chain: bool, /// Microsecond timestamp when this task finished (SWQOS returned); for per-SWQOS event→submit timing. submit_done_us: i64, @@ -295,7 +375,7 @@ impl ResultCollector { async fn wait_for_success( &self, - ) -> Option<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { + ) -> Option<(bool, Vec, Option, Vec)> { let start = Instant::now(); let timeout = std::time::Duration::from_secs(5); let poll_interval = std::time::Duration::from_millis(1000); @@ -307,7 +387,7 @@ impl ResultCollector { let mut submit_timings = Vec::new(); while let Some(result) = self.results.pop() { signatures.push(result.signature); - submit_timings.push((result.swqos_type, result.submit_done_us)); + submit_timings.push(result.submit_timing()); if result.success { has_success = true; } @@ -325,7 +405,7 @@ impl ResultCollector { let mut submit_timings = Vec::new(); while let Some(result) = self.results.pop() { signatures.push(result.signature); - submit_timings.push((result.swqos_type, result.submit_done_us)); + submit_timings.push(result.submit_timing()); // Prefer the error from the tx that actually landed if result.landed_on_chain && result.error.is_some() { landed_error = result.error; @@ -344,7 +424,7 @@ impl ResultCollector { let mut submit_timings = Vec::new(); while let Some(result) = self.results.pop() { signatures.push(result.signature); - submit_timings.push((result.swqos_type, result.submit_done_us)); + submit_timings.push(result.submit_timing()); if result.success { any_success = true; } @@ -367,7 +447,7 @@ impl ResultCollector { fn get_first( &self, - ) -> Option<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { + ) -> Option<(bool, Vec, Option, Vec)> { let mut signatures = Vec::new(); let mut has_success = false; let mut last_error = None; @@ -375,7 +455,7 @@ impl ResultCollector { while let Some(result) = self.results.pop() { signatures.push(result.signature); - submit_timings.push((result.swqos_type, result.submit_done_us)); + submit_timings.push(result.submit_timing()); if result.success { has_success = true; } @@ -396,7 +476,7 @@ impl ResultCollector { async fn wait_for_all_submitted( &self, timeout_secs: u64, - ) -> Option<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { + ) -> Option<(bool, Vec, Option, Vec)> { let start = Instant::now(); let primary = Duration::from_secs(timeout_secs); let poll_interval = Duration::from_millis(2); @@ -410,8 +490,7 @@ impl ResultCollector { // 若主窗口到时仍有未回包通道，晚到的 TaskResult 若立刻 drain 会丢签名——仅在该路径上拉长 grace。 let all_submitted = self.completed_count.load(Ordering::Acquire) >= self.total_tasks; if all_submitted { - // 全数已登记：仅留极短 settle，避免极端情况下最后一笔与计数可见性竞态。 - tokio::time::sleep(Duration::from_millis(35)).await; + tokio::task::yield_now().await; } else { tokio::time::sleep(Duration::from_millis(600)).await; while self.completed_count.load(Ordering::Acquire) < self.total_tasks { @@ -426,6 +505,63 @@ impl ResultCollector { } } +type GasFeeConfig = (SwqosType, GasFeeStrategyType, GasFeeStrategyValue); + +#[derive(Debug, Clone, Copy)] +struct SwqosTaskConfig { + task_ordinal: usize, + swqos_index: usize, + gas_fee_config: GasFeeConfig, +} + +impl TaskResult { + #[inline] + fn submit_timing(&self) -> SwqosSubmitTiming { + SwqosSubmitTiming { + swqos_type: self.swqos_type, + strategy_type: self.strategy_type, + submit_done_us: self.submit_done_us, + } + } +} + +fn select_swqos_task_configs( + swqos_types: &[SwqosType], + gas_fee_configs: &[GasFeeConfig], + with_tip: bool, + check_min_tip: bool, + min_tip_by_swqos: impl Fn(SwqosType) -> f64, +) -> Vec { + let mut task_configs = Vec::with_capacity(swqos_types.len() * 3); + for (i, swqos_type) in swqos_types.iter().copied().enumerate() { + if !with_tip && !matches!(swqos_type, SwqosType::Default) { + continue; + } + let check_tip = with_tip && !matches!(swqos_type, SwqosType::Default) && check_min_tip; + let min_tip = if check_tip { min_tip_by_swqos(swqos_type) } else { 0.0 }; + for config in gas_fee_configs { + if config.0 != swqos_type { + continue; + } + if check_tip && config.2.tip < min_tip { + if crate::common::sdk_log::sdk_log_enabled() { + println!( + "⚠️ Config filtered: {:?} tip {} is below minimum required {}", + config.0, config.2.tip, min_tip + ); + } + continue; + } + task_configs.push(SwqosTaskConfig { + task_ordinal: task_configs.len(), + swqos_index: i, + gas_fee_config: *config, + }); + } + } + task_configs +} + /// Execute trade on multiple SWQOS clients in parallel; returns success flag, all signatures, and last error. /// /// `sender_config` merges sender_thread_cores, effective_core_ids, max_sender_concurrency (precomputed at SDK init; no get_core_ids on hot path). @@ -445,7 +581,7 @@ pub async fn execute_parallel( use_dedicated_sender_threads: bool, sender_config: SenderConcurrencyConfig, check_min_tip: bool, -) -> Result<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { +) -> Result<(bool, Vec, Option, Vec)> { if swqos_clients.is_empty() { return Err(anyhow!("swqos_clients is empty")); } @@ -464,43 +600,32 @@ pub async fn execute_parallel( // One get_strategies call per batch (avoid N calls in loop). let gas_fee_configs = gas_fee_strategy.get_strategies(if is_buy { TradeType::Buy } else { TradeType::Sell }); - let mut task_configs = Vec::with_capacity(swqos_clients.len() * 3); - for (i, swqos_client) in swqos_clients.iter().enumerate() { - let swqos_type = swqos_client.get_swqos_type(); - if !with_tip && !matches!(swqos_type, SwqosType::Default) { - continue; - } - let check_tip = with_tip && !matches!(swqos_type, SwqosType::Default) && check_min_tip; - let min_tip = if check_tip { swqos_client.min_tip_sol() } else { 0.0 }; - for config in &gas_fee_configs { - if config.0 != swqos_type { - continue; - } - if check_tip { - if config.2.tip < min_tip && crate::common::sdk_log::sdk_log_enabled() { - println!( - "⚠️ Config filtered: {:?} tip {} is below minimum required {}", - config.0, config.2.tip, min_tip - ); - } - if config.2.tip < min_tip { - continue; - } - } - task_configs.push((i, swqos_client.clone(), *config)); - } - } + let swqos_types: Vec = + swqos_clients.iter().map(|swqos| swqos.get_swqos_type()).collect(); + let selected_task_configs = select_swqos_task_configs( + &swqos_types, + &gas_fee_configs, + with_tip, + check_min_tip, + |swqos_type| { + swqos_clients + .iter() + .find(|swqos| swqos.get_swqos_type() == swqos_type) + .map(|swqos| swqos.min_tip_sol()) + .unwrap_or(0.0) + }, + ); - if task_configs.is_empty() { + if selected_task_configs.is_empty() { return Err(anyhow!("No available gas fee strategy configs")); } - if is_buy && task_configs.len() > 1 && durable_nonce.is_none() { + if is_buy && selected_task_configs.len() > 1 && durable_nonce.is_none() { return Err(anyhow!("Multiple swqos transactions require durable_nonce to be set.",)); } // Task preparation completed: one shared context (clone once per batch), then minimal per-task data. - let channel_count = task_configs.len().max(1); + let channel_count = selected_task_configs.len().max(1); let collector = Arc::new(ResultCollector::new(channel_count)); // 上限最多 5s：单路卡死时才会等满；若所有通道在窗口内回完，主循环会提前结束（不睡满秒数）。 let submit_timeout_secs: u64 = @@ -534,10 +659,15 @@ pub async fn execute_parallel( let effective_core_ids = sender_config.effective_core_ids.as_slice(); let core_len = effective_core_ids.len().max(1); let mut tip_cache: FnvHashMap<*const (), Arc> = - FnvHashMap::with_capacity_and_hasher(task_configs.len(), BuildHasherDefault::default()); - for (i, swqos_client, gas_fee_strategy_config) in task_configs { - let core_id = effective_core_ids.get(i % core_len).copied(); + FnvHashMap::with_capacity_and_hasher( + selected_task_configs.len(), + BuildHasherDefault::default(), + ); + for task_config in selected_task_configs { + let swqos_client = swqos_clients[task_config.swqos_index].clone(); + let core_id = effective_core_ids.get(task_config.task_ordinal % core_len).copied(); let swqos_type = swqos_client.get_swqos_type(); + let gas_fee_strategy_config = task_config.gas_fee_config; let key = Arc::as_ptr(&swqos_client) as *const (); let tip_account = match tip_cache.get(&key) { Some(t) => t.clone(), @@ -561,10 +691,21 @@ pub async fn execute_parallel( tip_account, swqos_client, swqos_type, + strategy_type: gas_fee_strategy_config.1, core_id, use_affinity: !effective_core_ids.is_empty(), }; - let _ = queue.push(job); + if let Err(job) = queue.push(job) { + shared.collector.submit(TaskResult { + success: false, + signature: Signature::default(), + error: Some(anyhow!("SWQOS sender queue is full")), + swqos_type: job.swqos_type, + strategy_type: job.strategy_type, + landed_on_chain: false, + submit_done_us: crate::common::clock::now_micros(), + }); + } } } @@ -592,3 +733,73 @@ pub async fn execute_parallel( Err(anyhow!("All transactions failed")) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn value(cu_price: u64, tip: f64) -> GasFeeStrategyValue { + GasFeeStrategyValue { cu_limit: 100_000, cu_price, tip } + } + + #[test] + fn select_task_configs_keeps_two_fee_lanes_per_swqos() { + let swqos_types = [SwqosType::Jito, SwqosType::Helius]; + let configs = [ + (SwqosType::Jito, GasFeeStrategyType::LowTipHighCuPrice, value(400_000, 0.002)), + (SwqosType::Jito, GasFeeStrategyType::HighTipLowCuPrice, value(180_000, 0.005)), + (SwqosType::Helius, GasFeeStrategyType::LowTipHighCuPrice, value(400_000, 0.002)), + (SwqosType::Helius, GasFeeStrategyType::HighTipLowCuPrice, value(180_000, 0.005)), + ]; + + let selected = select_swqos_task_configs(&swqos_types, &configs, true, false, |_| 0.0); + + assert_eq!(selected.len(), 4); + assert_eq!( + selected.iter().filter(|task| task.gas_fee_config.0 == SwqosType::Jito).count(), + 2 + ); + assert_eq!( + selected.iter().filter(|task| task.gas_fee_config.0 == SwqosType::Helius).count(), + 2 + ); + assert_eq!( + selected.iter().map(|task| task.task_ordinal).collect::>(), + vec![0, 1, 2, 3] + ); + assert_eq!( + selected.iter().map(|task| task.swqos_index).collect::>(), + vec![0, 0, 1, 1] + ); + } + + #[test] + fn select_task_configs_applies_min_tip_per_lane() { + let swqos_types = [SwqosType::Jito]; + let configs = [ + (SwqosType::Jito, GasFeeStrategyType::LowTipHighCuPrice, value(400_000, 0.0001)), + (SwqosType::Jito, GasFeeStrategyType::HighTipLowCuPrice, value(180_000, 0.005)), + ]; + + let selected = select_swqos_task_configs(&swqos_types, &configs, true, true, |_| 0.001); + + assert_eq!(selected.len(), 1); + assert_eq!(selected[0].gas_fee_config.1, GasFeeStrategyType::HighTipLowCuPrice); + } + + #[test] + fn select_task_configs_without_tip_keeps_default_priority_fee_only() { + let swqos_types = [SwqosType::Jito, SwqosType::Default]; + let configs = [ + (SwqosType::Jito, GasFeeStrategyType::LowTipHighCuPrice, value(400_000, 0.002)), + (SwqosType::Default, GasFeeStrategyType::Normal, value(700_000, 0.0)), + ]; + + let selected = select_swqos_task_configs(&swqos_types, &configs, false, false, |_| 0.0); + + assert_eq!(selected.len(), 1); + assert_eq!(selected[0].gas_fee_config.0, SwqosType::Default); + assert_eq!(selected[0].gas_fee_config.2.cu_price, 700_000); + assert_eq!(selected[0].gas_fee_config.2.tip, 0.0); + } +} diff --git a/src/trading/core/executor.rs b/src/trading/core/executor.rs index 24584a5..5ecc226 100755 --- a/src/trading/core/executor.rs +++ b/src/trading/core/executor.rs @@ -12,10 +12,9 @@ use std::{ use tracing::{info, trace, warn}; use super::{params::SwapParams, traits::InstructionBuilder}; -use crate::swqos::SwqosType; use crate::swqos::TradeType; use crate::{ - common::{nonce_cache::DurableNonceInfo, GasFeeStrategy, SolanaRpcClient}, + common::{nonce_cache::DurableNonceInfo, GasFeeStrategy, SolanaRpcClient, SwqosSubmitTiming}, perf::syscall_bypass::SystemCallBypassManager, swqos::common::poll_any_transaction_confirmation, trading::core::{ @@ -56,7 +55,7 @@ impl TradeExecutor for GenericTradeExecutor { async fn swap( &self, params: SwapParams, - ) -> Result<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { + ) -> Result<(bool, Vec, Option, Vec)> { // Sample total start only when logging or simulate. 仅在有日志或 simulate 时取起点。 let total_start = (params.log_enabled || params.simulate).then(Instant::now); let timing_start_us: Option = if params.log_enabled { @@ -187,7 +186,7 @@ impl TradeExecutor for GenericTradeExecutor { Err(e) => (false, vec![], Some(anyhow::anyhow!("{}", e)), vec![]), }; // submit_timings 为完成先后顺序（先完成的先 push），打印不排序、不增加延迟 - let submit_timings_ref: &[(crate::swqos::SwqosType, i64)] = submit_timings.as_slice(); + let submit_timings_ref: &[SwqosSubmitTiming] = submit_timings.as_slice(); let result = if need_confirm { let confirm_result = if let Some(rpc) = params.rpc.as_ref() { @@ -257,7 +256,7 @@ async fn simulate_transaction( is_buy: bool, with_tip: bool, gas_fee_strategy: GasFeeStrategy, -) -> Result<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { +) -> Result<(bool, Vec, Option, Vec)> { use crate::trading::common::build_transaction; use solana_client::rpc_config::RpcSimulateTransactionConfig; use solana_commitment_config::CommitmentLevel; @@ -351,6 +350,7 @@ async fn simulate_transaction( #[cfg(test)] mod tests { + use crate::common::GasFeeStrategyType; use crate::swqos::SwqosType; /// 运行 `cargo test -p sol-trade-sdk log_timing_preview -- --nocapture` 查看日志打印效果 @@ -377,15 +377,17 @@ mod tests { ); println!("\n--- 2. 每个 SWQOS 独立耗时：submit_done=起点→该通道提交完成, confirmed=该通道提交→链上确认, total=起点→链上确认 ---\n"); - for (swqos_type, submit_ms, confirmed_ms, total_ms) in [ - (SwqosType::Jito, 45.12, 83.38, 128.50), - (SwqosType::Helius, 52.30, 76.20, 128.50), - (SwqosType::ZeroSlot, 48.90, 79.60, 128.50), + for (swqos_type, strategy_type, submit_ms, confirmed_ms, total_ms) in [ + (SwqosType::Jito, GasFeeStrategyType::LowTipHighCuPrice, 45.12, 83.38, 128.50), + (SwqosType::Jito, GasFeeStrategyType::HighTipLowCuPrice, 46.08, 82.42, 128.50), + (SwqosType::Helius, GasFeeStrategyType::LowTipHighCuPrice, 52.30, 76.20, 128.50), + (SwqosType::ZeroSlot, GasFeeStrategyType::Normal, 48.90, 79.60, 128.50), ] { println!( - " [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", + " [SDK][{:width$}] {} {} submit_done: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", swqos_type.as_str(), dir, + strategy_type.as_str(), submit_ms, confirmed_ms, total_ms, @@ -396,13 +398,16 @@ mod tests { println!( "\n--- 3. 不等待链上确认时：每行 total = 该通道 submit_done（提交完成总耗时）---\n" ); - for (swqos_type, submit_ms, total_ms) in - [(SwqosType::Jito, 44.20, 44.20), (SwqosType::Helius, 51.80, 51.80)] - { + for (swqos_type, strategy_type, submit_ms, total_ms) in [ + (SwqosType::Jito, GasFeeStrategyType::LowTipHighCuPrice, 44.20, 44.20), + (SwqosType::Jito, GasFeeStrategyType::HighTipLowCuPrice, 45.10, 45.10), + (SwqosType::Helius, GasFeeStrategyType::Normal, 51.80, 51.80), + ] { println!( - " [SDK][{:width$}] {} submit_done: {:.4} ms, confirmed: -, total: {:.4} ms", + " [SDK][{:width$}] {} {} submit_done: {:.4} ms, confirmed: -, total: {:.4} ms", swqos_type.as_str(), dir, + strategy_type.as_str(), submit_ms, total_ms, width = w diff --git a/src/trading/core/traits.rs b/src/trading/core/traits.rs index 7764d8b..1b02d5e 100755 --- a/src/trading/core/traits.rs +++ b/src/trading/core/traits.rs @@ -1,4 +1,4 @@ -use crate::swqos::SwqosType; +use crate::common::SwqosSubmitTiming; use crate::trading::SwapParams; use anyhow::Result; use solana_sdk::{instruction::Instruction, signature::Signature}; @@ -12,7 +12,7 @@ pub trait TradeExecutor: Send + Sync { async fn swap( &self, params: SwapParams, - ) -> Result<(bool, Vec, Option, Vec<(SwqosType, i64)>)>; + ) -> Result<(bool, Vec, Option, Vec)>; /// 获取协议名称 fn protocol_name(&self) -> &'static str; }