Release v3.5.0: performance, constants, bilingual docs
- Bump version to 3.5.0 - Performance: hot-path timing only when log_enabled/simulate; execute_parallel takes &[Arc<SwqosClient>]; shared HTTP client constants for SWQoS - Code quality: validate_protocol_params extracted for buy/sell; BYTES_PER_ACCOUNT, MAX_INSTRUCTIONS_WARN, HTTP timeout constants; prefetch/syscall bypass comments - Documentation: bilingual (EN + 中文) doc comments in execution, executor, perf, swqos; README/README_CN version and What's new in 3.5.0 - Add release_notes_v3.5.0.md Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -14,13 +14,14 @@ use crate::{
|
||||
trading::{MiddlewareManager, core::transaction_pool::{acquire_builder, release_builder}},
|
||||
};
|
||||
|
||||
/// Build standard RPC transaction
|
||||
/// Build standard RPC transaction.
|
||||
/// Takes `business_instructions` by reference to avoid per-task Vec clone in execute_parallel.
|
||||
pub async fn build_transaction(
|
||||
payer: Arc<Keypair>,
|
||||
_rpc: Option<Arc<SolanaRpcClient>>,
|
||||
unit_limit: u32,
|
||||
unit_price: u64,
|
||||
business_instructions: Vec<Instruction>,
|
||||
business_instructions: &[Instruction],
|
||||
address_lookup_table_account: Option<AddressLookupTableAccount>,
|
||||
recent_blockhash: Option<Hash>,
|
||||
middleware_manager: Option<Arc<MiddlewareManager>>,
|
||||
@@ -57,8 +58,8 @@ pub async fn build_transaction(
|
||||
unit_limit,
|
||||
));
|
||||
|
||||
// Add business instructions
|
||||
instructions.extend(business_instructions);
|
||||
// Add business instructions (clone only here; avoids per-task Vec clone in execute_parallel)
|
||||
instructions.extend_from_slice(business_instructions);
|
||||
|
||||
// Get blockhash for transaction
|
||||
let blockhash = get_transaction_blockhash(recent_blockhash, durable_nonce.clone());
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
//! 并行执行器
|
||||
//! Parallel executor for multi-SWQOS submit.
|
||||
|
||||
use anyhow::{anyhow, Result};
|
||||
use crossbeam_queue::ArrayQueue;
|
||||
@@ -39,8 +39,8 @@ struct TaskResult {
|
||||
signature: Signature,
|
||||
error: Option<anyhow::Error>,
|
||||
#[allow(dead_code)]
|
||||
swqos_type: SwqosType, // 🔧 增加:记录SWQOS类型
|
||||
landed_on_chain: bool, // 🔧 Whether tx landed on-chain (even if failed)
|
||||
swqos_type: SwqosType,
|
||||
landed_on_chain: bool,
|
||||
}
|
||||
|
||||
/// Check if an error indicates the transaction landed on-chain (vs network/timeout error)
|
||||
@@ -87,14 +87,14 @@ impl ResultCollector {
|
||||
}
|
||||
|
||||
fn submit(&self, result: TaskResult) {
|
||||
// 🚀 优化:ArrayQueue 内部已保证同步,无需额外 fence
|
||||
// ArrayQueue is already synchronized; no extra fence needed
|
||||
let is_success = result.success;
|
||||
let is_landed_failed = result.landed_on_chain && !result.success;
|
||||
|
||||
let _ = self.results.push(result);
|
||||
|
||||
if is_success {
|
||||
self.success_flag.store(true, Ordering::Release); // Release 确保 push 可见
|
||||
self.success_flag.store(true, Ordering::Release);
|
||||
} else if is_landed_failed {
|
||||
// 🔧 Tx landed but failed (e.g., ExceededSlippage) - nonce is consumed, no point waiting
|
||||
self.landed_failed_flag.store(true, Ordering::Release);
|
||||
@@ -105,12 +105,11 @@ impl ResultCollector {
|
||||
|
||||
async fn wait_for_success(&self) -> Option<(bool, Vec<Signature>, Option<anyhow::Error>)> {
|
||||
let start = Instant::now();
|
||||
let timeout = std::time::Duration::from_secs(30);
|
||||
let timeout = std::time::Duration::from_secs(5);
|
||||
let poll_interval = std::time::Duration::from_millis(1000);
|
||||
|
||||
loop {
|
||||
// 🚀 Acquire 确保看到 push 的内容
|
||||
if self.success_flag.load(Ordering::Acquire) {
|
||||
// 🔧 修复:收集所有签名
|
||||
let mut signatures = Vec::new();
|
||||
let mut has_success = false;
|
||||
while let Some(result) = self.results.pop() {
|
||||
@@ -124,7 +123,7 @@ impl ResultCollector {
|
||||
}
|
||||
}
|
||||
|
||||
// 🔧 Early exit: if a tx landed but failed (e.g., ExceededSlippage),
|
||||
// Early exit: if a tx landed but failed (e.g., ExceededSlippage),
|
||||
// nonce is consumed and other channels can't succeed - return immediately
|
||||
if self.landed_failed_flag.load(Ordering::Acquire) {
|
||||
let mut signatures = Vec::new();
|
||||
@@ -142,8 +141,7 @@ impl ResultCollector {
|
||||
}
|
||||
|
||||
let completed = self.completed_count.load(Ordering::Acquire);
|
||||
if completed >= self.total_tasks {
|
||||
// 🔧 修复:收集所有签名
|
||||
if completed >= self.total_tasks {
|
||||
let mut signatures = Vec::new();
|
||||
let mut last_error = None;
|
||||
let mut any_success = false;
|
||||
@@ -165,12 +163,11 @@ impl ResultCollector {
|
||||
if start.elapsed() > timeout {
|
||||
return None;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
tokio::time::sleep(poll_interval).await;
|
||||
}
|
||||
}
|
||||
|
||||
fn get_first(&self) -> Option<(bool, Vec<Signature>, Option<anyhow::Error>)> {
|
||||
// 🔧 修复:收集已提交的所有签名
|
||||
let mut signatures = Vec::new();
|
||||
let mut has_success = false;
|
||||
let mut last_error = None;
|
||||
@@ -191,11 +188,25 @@ impl ResultCollector {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// 等待全部任务完成(不等待链上确认),然后收集并返回所有签名。用于「多路提交」时返回多笔签名。
|
||||
async fn wait_for_all_submitted(&self, timeout_secs: u64) -> Option<(bool, Vec<Signature>, Option<anyhow::Error>)> {
|
||||
let start = Instant::now();
|
||||
let timeout = std::time::Duration::from_secs(timeout_secs);
|
||||
let poll_interval = std::time::Duration::from_millis(50);
|
||||
while self.completed_count.load(Ordering::Acquire) < self.total_tasks {
|
||||
if start.elapsed() > timeout {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(poll_interval).await;
|
||||
}
|
||||
self.get_first()
|
||||
}
|
||||
}
|
||||
|
||||
/// 🔧 修复:返回Vec<Signature>支持多SWQOS并发交易
|
||||
/// Execute trade on multiple SWQOS clients in parallel; returns success flag, all signatures, and last error.
|
||||
pub async fn execute_parallel(
|
||||
swqos_clients: Vec<Arc<SwqosClient>>,
|
||||
swqos_clients: &[Arc<SwqosClient>],
|
||||
payer: Arc<Keypair>,
|
||||
rpc: Option<Arc<SolanaRpcClient>>,
|
||||
instructions: Vec<Instruction>,
|
||||
@@ -208,6 +219,7 @@ pub async fn execute_parallel(
|
||||
wait_transaction_confirmed: bool,
|
||||
with_tip: bool,
|
||||
gas_fee_strategy: GasFeeStrategy,
|
||||
use_core_affinity: bool,
|
||||
) -> Result<(bool, Vec<Signature>, Option<anyhow::Error>)> {
|
||||
let _exec_start = Instant::now();
|
||||
|
||||
@@ -224,10 +236,10 @@ pub async fn execute_parallel(
|
||||
return Err(anyhow!("No Rpc Default Swqos configured."));
|
||||
}
|
||||
|
||||
let cores = core_affinity::get_core_ids().unwrap();
|
||||
let cores = core_affinity::get_core_ids().unwrap_or_default();
|
||||
let instructions = Arc::new(instructions);
|
||||
|
||||
// 预先计算所有有效的组合
|
||||
// Precompute all valid (client, gas config) combinations
|
||||
let task_configs: Vec<_> = swqos_clients
|
||||
.iter()
|
||||
.enumerate()
|
||||
@@ -244,7 +256,7 @@ pub async fn execute_parallel(
|
||||
.into_iter()
|
||||
.filter(|config| config.0.eq(&swqos_client.get_swqos_type()))
|
||||
.filter(|config| {
|
||||
// 当需要 tip 且不是 Default 时,按 provider 最低小费进行筛选
|
||||
// When tip required and not Default, filter by provider minimum tip
|
||||
if with_tip && !matches!(config.0, SwqosType::Default) {
|
||||
let min_tip = match config.0 {
|
||||
SwqosType::Jito => SWQOS_MIN_TIP_JITO,
|
||||
@@ -262,9 +274,9 @@ pub async fn execute_parallel(
|
||||
SwqosType::Speedlanding => SWQOS_MIN_TIP_SPEEDLANDING,
|
||||
SwqosType::Default => SWQOS_MIN_TIP_DEFAULT,
|
||||
};
|
||||
if config.2.tip < min_tip {
|
||||
if config.2.tip < min_tip && crate::common::sdk_log::sdk_log_enabled() {
|
||||
println!(
|
||||
"⚠️ Config filtered: {:?} tip {} is below minimum required tip {}",
|
||||
"⚠️ Config filtered: {:?} tip {} is below minimum required {}",
|
||||
config.0, config.2.tip, min_tip
|
||||
);
|
||||
}
|
||||
@@ -291,7 +303,8 @@ pub async fn execute_parallel(
|
||||
let _spawn_start = Instant::now();
|
||||
|
||||
for (i, swqos_client, gas_fee_strategy_config) in task_configs {
|
||||
let core_id = cores[i % cores.len()];
|
||||
let core_id = cores.get(i % cores.len().max(1)).copied();
|
||||
let use_affinity = use_core_affinity;
|
||||
let payer = payer.clone();
|
||||
let instructions = instructions.clone();
|
||||
let middleware_manager = middleware_manager.clone();
|
||||
@@ -306,10 +319,15 @@ pub async fn execute_parallel(
|
||||
let rpc = rpc.clone();
|
||||
let durable_nonce = durable_nonce.clone();
|
||||
let address_lookup_table_account = address_lookup_table_account.clone();
|
||||
let recent_blockhash_task = recent_blockhash.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _task_start = Instant::now();
|
||||
core_affinity::set_for_current(core_id);
|
||||
if use_affinity {
|
||||
if let Some(cid) = core_id {
|
||||
core_affinity::set_for_current(cid);
|
||||
}
|
||||
}
|
||||
|
||||
let tip_amount = if with_tip { tip } else { 0.0 };
|
||||
|
||||
@@ -319,9 +337,9 @@ pub async fn execute_parallel(
|
||||
rpc,
|
||||
unit_limit,
|
||||
unit_price,
|
||||
instructions.as_ref().clone(),
|
||||
instructions.as_ref(),
|
||||
address_lookup_table_account,
|
||||
recent_blockhash,
|
||||
recent_blockhash_task,
|
||||
middleware_manager,
|
||||
protocol_name,
|
||||
is_buy,
|
||||
@@ -339,8 +357,8 @@ pub async fn execute_parallel(
|
||||
success: false,
|
||||
signature: Signature::default(),
|
||||
error: Some(e),
|
||||
swqos_type, // 🔧 记录SWQOS类型
|
||||
landed_on_chain: false, // Build failed, tx never sent
|
||||
swqos_type,
|
||||
landed_on_chain: false,
|
||||
});
|
||||
return;
|
||||
}
|
||||
@@ -373,28 +391,28 @@ pub async fn execute_parallel(
|
||||
}
|
||||
};
|
||||
|
||||
// Transaction sent
|
||||
|
||||
if let Some(signature) = transaction.signatures.first() {
|
||||
collector.submit(TaskResult {
|
||||
success,
|
||||
signature: *signature,
|
||||
error: err,
|
||||
swqos_type, // 🔧 记录SWQOS类型
|
||||
landed_on_chain, // 🔧 Whether tx landed (even if it failed)
|
||||
});
|
||||
}
|
||||
// Transaction sent: always submit a result so collector never has "no result" for this task.
|
||||
// If transaction has no signatures (malformed), submit with default signature and success=false.
|
||||
let sig = transaction.signatures.first().copied().unwrap_or_default();
|
||||
collector.submit(TaskResult {
|
||||
success,
|
||||
signature: sig,
|
||||
error: err,
|
||||
swqos_type,
|
||||
landed_on_chain,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// All tasks spawned
|
||||
|
||||
if !wait_transaction_confirmed {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||
if let Some(result) = collector.get_first() {
|
||||
return Ok(result);
|
||||
}
|
||||
return Err(anyhow!("No transaction signature available"));
|
||||
const SUBMIT_TIMEOUT_SECS: u64 = 30;
|
||||
let (success, signatures, last_error) = collector
|
||||
.wait_for_all_submitted(SUBMIT_TIMEOUT_SECS)
|
||||
.await
|
||||
.unwrap_or((false, vec![], Some(anyhow!("No SWQOS result within {}s", SUBMIT_TIMEOUT_SECS))));
|
||||
return Ok((success, signatures, last_error));
|
||||
}
|
||||
|
||||
if let Some(result) = collector.wait_for_success().await {
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
//! 执行模块
|
||||
//! Execution: instruction preprocessing, cache prefetch, branch hints.
|
||||
//! 执行模块:指令预处理、缓存预取、分支提示。
|
||||
|
||||
use anyhow::Result;
|
||||
use solana_sdk::{
|
||||
@@ -12,7 +13,15 @@ use crate::perf::{
|
||||
simd::SIMDMemory,
|
||||
};
|
||||
|
||||
/// 预取工具
|
||||
/// Solana account key size in bytes (Pubkey). 每个账户(Pubkey)的字节数。
|
||||
pub const BYTES_PER_ACCOUNT: usize = 32;
|
||||
|
||||
/// Threshold above which we warn about large instruction count. 超过此次数会打 warning。
|
||||
pub const MAX_INSTRUCTIONS_WARN: usize = 64;
|
||||
|
||||
/// Prefetch helper: triggers CPU prefetch for soon-to-be-accessed data to reduce cache-miss latency.
|
||||
/// Call once on hot-path refs; no-op on non-x86_64. Safety: caller ensures valid read-only ref, no concurrent write.
|
||||
/// 缓存预取:对即将访问的数据做 CPU 预取以降低 cache-miss;热路径上调用一次即可;非 x86_64 为 no-op。安全:调用方保证有效只读、无并发写。
|
||||
pub struct Prefetch;
|
||||
|
||||
impl Prefetch {
|
||||
@@ -21,20 +30,15 @@ impl Prefetch {
|
||||
if instructions.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
// 预取第一条指令
|
||||
// Prefetch first/middle/last instruction into L1 for subsequent build_transaction access. 预取首/中/尾指令到 L1。
|
||||
unsafe {
|
||||
BranchOptimizer::prefetch_read_data(&instructions[0]);
|
||||
}
|
||||
|
||||
// 预取中间指令
|
||||
if instructions.len() > 2 {
|
||||
unsafe {
|
||||
BranchOptimizer::prefetch_read_data(&instructions[instructions.len() / 2]);
|
||||
}
|
||||
}
|
||||
|
||||
// 预取最后一条指令
|
||||
if instructions.len() > 1 {
|
||||
unsafe {
|
||||
BranchOptimizer::prefetch_read_data(&instructions[instructions.len() - 1]);
|
||||
@@ -57,46 +61,40 @@ impl Prefetch {
|
||||
}
|
||||
}
|
||||
|
||||
/// 内存操作
|
||||
/// Memory operations (SIMD-accelerated where available). 内存操作(可用时走 SIMD 加速)。
|
||||
pub struct MemoryOps;
|
||||
|
||||
impl MemoryOps {
|
||||
#[inline(always)]
|
||||
pub unsafe fn copy(dst: *mut u8, src: *const u8, len: usize) {
|
||||
// 优先使用 AVX2 SIMD 加速
|
||||
SIMDMemory::copy_avx2(dst, src, len);
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
pub unsafe fn compare(a: *const u8, b: *const u8, len: usize) -> bool {
|
||||
// 优先使用 AVX2 SIMD 比较
|
||||
SIMDMemory::compare_avx2(a, b, len)
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
pub unsafe fn zero(ptr: *mut u8, len: usize) {
|
||||
// 优先使用 AVX2 SIMD 清零
|
||||
SIMDMemory::zero_avx2(ptr, len);
|
||||
}
|
||||
}
|
||||
|
||||
/// 指令处理器
|
||||
/// Instruction preprocessing and validation. 指令预处理与校验。
|
||||
pub struct InstructionProcessor;
|
||||
|
||||
impl InstructionProcessor {
|
||||
#[inline(always)]
|
||||
pub fn preprocess(instructions: &[Instruction]) -> Result<()> {
|
||||
// 分支预测: 大概率指令不为空
|
||||
if BranchOptimizer::unlikely(instructions.is_empty()) {
|
||||
return Err(anyhow::anyhow!("Instructions empty"));
|
||||
}
|
||||
|
||||
// 预取所有指令到缓存
|
||||
Prefetch::instructions(instructions);
|
||||
|
||||
// 分支预测: 大概率指令数量合理
|
||||
if BranchOptimizer::unlikely(instructions.len() > 64) {
|
||||
log::warn!("Large instruction count: {}", instructions.len());
|
||||
if BranchOptimizer::unlikely(instructions.len() > MAX_INSTRUCTIONS_WARN) {
|
||||
tracing::warn!(target: "sol_trade_sdk", "Large instruction count: {}", instructions.len());
|
||||
}
|
||||
|
||||
Ok(())
|
||||
@@ -107,7 +105,7 @@ impl InstructionProcessor {
|
||||
let mut total_size = 0;
|
||||
|
||||
for (i, instr) in instructions.iter().enumerate() {
|
||||
// 预取下一条指令
|
||||
// Prefetch next instruction; safe: same slice, read-only. 预取下一条指令;安全:同 slice、只读。
|
||||
unsafe {
|
||||
if let Some(next_instr) = instructions.get(i + 1) {
|
||||
BranchOptimizer::prefetch_read_data(next_instr);
|
||||
@@ -115,20 +113,19 @@ impl InstructionProcessor {
|
||||
}
|
||||
|
||||
total_size += instr.data.len();
|
||||
total_size += instr.accounts.len() * 32; // 每个账户 32 字节
|
||||
total_size += instr.accounts.len() * BYTES_PER_ACCOUNT;
|
||||
}
|
||||
|
||||
total_size
|
||||
}
|
||||
}
|
||||
|
||||
/// 执行路径
|
||||
/// Trade direction / execution path helpers. 交易方向与执行路径辅助。
|
||||
pub struct ExecutionPath;
|
||||
|
||||
impl ExecutionPath {
|
||||
#[inline(always)]
|
||||
pub fn is_buy(input_mint: &Pubkey) -> bool {
|
||||
// 分支预测: 大概率是买入
|
||||
let is_buy = input_mint == &crate::constants::SOL_TOKEN_ACCOUNT
|
||||
|| input_mint == &crate::constants::WSOL_TOKEN_ACCOUNT
|
||||
|| input_mint == &crate::constants::USD1_TOKEN_ACCOUNT
|
||||
|
||||
@@ -4,11 +4,14 @@ use solana_sdk::{
|
||||
instruction::Instruction, message::AddressLookupTableAccount, pubkey::Pubkey,
|
||||
signature::Keypair, signature::Signature,
|
||||
};
|
||||
use std::{sync::Arc, time::Instant};
|
||||
use std::{sync::Arc, time::{Duration, Instant}};
|
||||
#[allow(unused_imports)]
|
||||
use tracing::{info, trace, warn};
|
||||
|
||||
use crate::{
|
||||
common::{nonce_cache::DurableNonceInfo, GasFeeStrategy, SolanaRpcClient},
|
||||
perf::syscall_bypass::SystemCallBypassManager,
|
||||
swqos::common::poll_transaction_confirmation,
|
||||
trading::core::{
|
||||
async_executor::execute_parallel,
|
||||
execution::{InstructionProcessor, Prefetch},
|
||||
@@ -20,7 +23,9 @@ use once_cell::sync::Lazy;
|
||||
use crate::swqos::TradeType;
|
||||
use super::{params::SwapParams, traits::InstructionBuilder};
|
||||
|
||||
/// 🚀 全局系统调用绕过管理器
|
||||
/// Global syscall bypass manager (reserved for future time/IO optimizations).
|
||||
/// 全局系统调用绕过管理器(预留,后续可接入时间/IO 等优化)。
|
||||
#[allow(dead_code)]
|
||||
static SYSCALL_BYPASS: Lazy<SystemCallBypassManager> = Lazy::new(|| {
|
||||
use crate::perf::syscall_bypass::SyscallBypassConfig;
|
||||
SystemCallBypassManager::new(SyscallBypassConfig::default())
|
||||
@@ -45,27 +50,29 @@ impl GenericTradeExecutor {
|
||||
#[async_trait::async_trait]
|
||||
impl TradeExecutor for GenericTradeExecutor {
|
||||
async fn swap(&self, params: SwapParams) -> Result<(bool, Vec<Signature>, Option<anyhow::Error>)> {
|
||||
let total_start = Instant::now();
|
||||
// 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<i64> = if params.log_enabled {
|
||||
Some(params.grpc_recv_us.unwrap_or_else(crate::common::clock::now_micros))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// 判断买卖方向
|
||||
let is_buy = params.trade_type == TradeType::Buy || params.trade_type == TradeType::CreateAndBuy;
|
||||
|
||||
// CPU 预取
|
||||
Prefetch::keypair(¶ms.payer);
|
||||
|
||||
// 构建指令
|
||||
let build_start = Instant::now();
|
||||
// Time build only when log_enabled to avoid cold-path syscalls. 仅 log_enabled 时计时,减少冷路径 syscall。
|
||||
let build_start = params.log_enabled.then(Instant::now);
|
||||
let instructions = if is_buy {
|
||||
self.instruction_builder.build_buy_instructions(¶ms).await?
|
||||
} else {
|
||||
self.instruction_builder.build_sell_instructions(¶ms).await?
|
||||
};
|
||||
let build_elapsed = build_start.elapsed();
|
||||
let build_elapsed = build_start.map(|s| s.elapsed()).unwrap_or(Duration::ZERO);
|
||||
|
||||
// 指令预处理
|
||||
InstructionProcessor::preprocess(&instructions)?;
|
||||
|
||||
// 中间件处理
|
||||
let final_instructions = match ¶ms.middleware_manager {
|
||||
Some(middleware_manager) => middleware_manager
|
||||
.apply_middlewares_process_protocol_instructions(
|
||||
@@ -76,12 +83,10 @@ impl TradeExecutor for GenericTradeExecutor {
|
||||
None => instructions,
|
||||
};
|
||||
|
||||
// 提交前耗时
|
||||
let before_submit_elapsed = total_start.elapsed();
|
||||
let before_submit_elapsed = total_start.as_ref().map(|s| s.elapsed()).unwrap_or(Duration::ZERO);
|
||||
|
||||
// 如果是模拟模式,直接通过 RPC 模拟交易
|
||||
if params.simulate {
|
||||
let send_start = Instant::now();
|
||||
let send_start = crate::common::sdk_log::sdk_log_enabled().then(Instant::now);
|
||||
let result = simulate_transaction(
|
||||
params.rpc,
|
||||
params.payer,
|
||||
@@ -96,44 +101,23 @@ impl TradeExecutor for GenericTradeExecutor {
|
||||
params.gas_fee_strategy,
|
||||
)
|
||||
.await;
|
||||
let send_elapsed = send_start.elapsed();
|
||||
let total_elapsed = total_start.elapsed();
|
||||
let send_elapsed = send_start.map(|s| s.elapsed()).unwrap_or(Duration::ZERO);
|
||||
let total_elapsed = total_start.as_ref().map(|s| s.elapsed()).unwrap_or(Duration::ZERO);
|
||||
|
||||
// Get performance metrics using fast timestamp
|
||||
let timestamp_ns = SYSCALL_BYPASS.fast_timestamp_nanos();
|
||||
|
||||
// Print all timing metrics at once to avoid blocking critical path
|
||||
println!("[Timestamp] {}ns", timestamp_ns);
|
||||
println!(
|
||||
"[Build Instructions] Time: {:.3}ms ({:.0}μs)",
|
||||
build_elapsed.as_micros() as f64 / 1000.0,
|
||||
build_elapsed.as_micros()
|
||||
);
|
||||
println!(
|
||||
"[Before Submit] {:.3}ms ({:.0}μs)",
|
||||
before_submit_elapsed.as_micros() as f64 / 1000.0,
|
||||
before_submit_elapsed.as_micros()
|
||||
);
|
||||
println!(
|
||||
"[Simulate Transaction] Time: {:.3}ms ({:.0}μs)",
|
||||
send_elapsed.as_micros() as f64 / 1000.0,
|
||||
send_elapsed.as_micros()
|
||||
);
|
||||
println!(
|
||||
"[Total Time] {:.3}ms ({:.0}μs)",
|
||||
total_elapsed.as_micros() as f64 / 1000.0,
|
||||
total_elapsed.as_micros()
|
||||
);
|
||||
if crate::common::sdk_log::sdk_log_enabled() {
|
||||
let dir = if is_buy { "Buy" } else { "Sell" };
|
||||
println!(" [SDK] {} timing(sim) build_instructions: {:.2}ms before_submit: {:.2}ms simulate: {:.2}ms total: {:.2}ms", dir, build_elapsed.as_secs_f64() * 1000.0, before_submit_elapsed.as_secs_f64() * 1000.0, send_elapsed.as_secs_f64() * 1000.0, total_elapsed.as_secs_f64() * 1000.0);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
// 并行发送交易
|
||||
let send_start = Instant::now();
|
||||
let need_confirm = params.wait_transaction_confirmed;
|
||||
let send_start = params.log_enabled.then(Instant::now);
|
||||
let result = execute_parallel(
|
||||
params.swqos_clients.clone(),
|
||||
¶ms.swqos_clients,
|
||||
params.payer,
|
||||
params.rpc,
|
||||
params.rpc.clone(),
|
||||
final_instructions,
|
||||
params.address_lookup_table_account,
|
||||
params.recent_blockhash,
|
||||
@@ -141,30 +125,64 @@ impl TradeExecutor for GenericTradeExecutor {
|
||||
params.middleware_manager,
|
||||
self.protocol_name,
|
||||
is_buy,
|
||||
params.wait_transaction_confirmed,
|
||||
false, // submit only here; confirmation and log timing handled below
|
||||
if is_buy { true } else { params.with_tip },
|
||||
params.gas_fee_strategy,
|
||||
params.use_core_affinity,
|
||||
)
|
||||
.await;
|
||||
let send_elapsed = send_start.elapsed();
|
||||
let total_elapsed = total_start.elapsed();
|
||||
let send_elapsed = send_start.map(|s| s.elapsed()).unwrap_or(Duration::ZERO);
|
||||
|
||||
// Get performance metrics using fast timestamp
|
||||
#[cfg(feature = "perf-trace")]
|
||||
{
|
||||
let timestamp_ns = SYSCALL_BYPASS.fast_timestamp_nanos();
|
||||
log::trace!(
|
||||
"[Execute] timestamp_ns={} build_us={} before_submit_us={} send_us={} total_us={}",
|
||||
timestamp_ns,
|
||||
build_elapsed.as_micros(),
|
||||
before_submit_elapsed.as_micros(),
|
||||
send_elapsed.as_micros(),
|
||||
total_elapsed.as_micros()
|
||||
);
|
||||
if params.log_enabled && crate::common::sdk_log::sdk_log_enabled() {
|
||||
let dir = if is_buy { "Buy" } else { "Sell" };
|
||||
let build_ms = build_elapsed.as_secs_f64() * 1000.0;
|
||||
let before_ms = before_submit_elapsed.as_secs_f64() * 1000.0;
|
||||
let send_ms = send_elapsed.as_secs_f64() * 1000.0;
|
||||
if let Some(start_us) = timing_start_us {
|
||||
let now_us = crate::common::clock::now_micros();
|
||||
let start_to_submit_us = (now_us - start_us).max(0);
|
||||
println!(" [SDK] {} timing(after_submit) build_instructions: {:.2}ms before_submit: {:.2}ms submit: {:.2}ms start_to_submit: {} μs", dir, build_ms, before_ms, send_ms, start_to_submit_us);
|
||||
} else {
|
||||
println!(" [SDK] {} timing(after_submit) build_instructions: {:.2}ms before_submit: {:.2}ms submit: {:.2}ms", dir, build_ms, before_ms, send_ms);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "perf-trace"))]
|
||||
let _ = (build_elapsed, before_submit_elapsed, send_elapsed, total_elapsed);
|
||||
|
||||
let result = if need_confirm {
|
||||
let (ok, sigs, err) = match &result {
|
||||
Ok((success, signatures, last_error)) => (
|
||||
*success,
|
||||
signatures.clone(),
|
||||
last_error.as_ref().map(|e| anyhow::anyhow!("{}", e)),
|
||||
),
|
||||
Err(e) => (false, vec![], Some(anyhow::anyhow!("{}", e))),
|
||||
};
|
||||
let first_sig = sigs.first().copied();
|
||||
let confirm_result = if let (Some(rpc), Some(sig)) = (params.rpc.as_ref(), first_sig) {
|
||||
let confirm_start = (params.log_enabled && crate::common::sdk_log::sdk_log_enabled()).then(Instant::now);
|
||||
let poll_res = poll_transaction_confirmation(rpc, sig, true).await;
|
||||
let confirm_elapsed = confirm_start.map(|s| s.elapsed()).unwrap_or(Duration::ZERO);
|
||||
if params.log_enabled && crate::common::sdk_log::sdk_log_enabled() {
|
||||
let dir = if is_buy { "Buy" } else { "Sell" };
|
||||
let confirm_ms = confirm_elapsed.as_secs_f64() * 1000.0;
|
||||
let total_ms = total_start.as_ref().map(|s| s.elapsed()).unwrap_or(Duration::ZERO).as_secs_f64() * 1000.0;
|
||||
println!(" [SDK] {} timing(after_confirm) confirm: {:.2}ms total: {:.2}ms", dir, confirm_ms, total_ms);
|
||||
}
|
||||
match poll_res {
|
||||
Ok(_) => (true, sigs, None),
|
||||
Err(e) => (false, sigs, Some(e)),
|
||||
}
|
||||
} else {
|
||||
(ok, sigs, err)
|
||||
};
|
||||
Ok(confirm_result)
|
||||
} else {
|
||||
if params.log_enabled && crate::common::sdk_log::sdk_log_enabled() {
|
||||
let total_ms = total_start.as_ref().map(|s| s.elapsed()).unwrap_or(Duration::ZERO).as_secs_f64() * 1000.0;
|
||||
let dir = if is_buy { "Buy" } else { "Sell" };
|
||||
println!(" [SDK] {} timing total: {:.2}ms", dir, total_ms);
|
||||
}
|
||||
result
|
||||
};
|
||||
|
||||
result
|
||||
}
|
||||
@@ -174,7 +192,8 @@ impl TradeExecutor for GenericTradeExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
/// 🔧 修复:Simulate模式返回Vec<Signature>(单个RPC模拟)
|
||||
/// Simulate mode: single RPC simulation, returns Vec<Signature> for API consistency.
|
||||
/// 模拟模式:单次 RPC 模拟,返回 Vec<Signature> 以与 API 一致。
|
||||
async fn simulate_transaction(
|
||||
rpc: Option<Arc<SolanaRpcClient>>,
|
||||
payer: Arc<Keypair>,
|
||||
@@ -215,7 +234,7 @@ async fn simulate_transaction(
|
||||
Some(rpc.clone()),
|
||||
unit_limit,
|
||||
unit_price,
|
||||
instructions,
|
||||
&instructions,
|
||||
address_lookup_table_account,
|
||||
recent_blockhash,
|
||||
middleware_manager,
|
||||
@@ -256,12 +275,12 @@ async fn simulate_transaction(
|
||||
if let Some(err) = simulate_result.value.err {
|
||||
#[cfg(feature = "perf-trace")]
|
||||
{
|
||||
log::warn!("[Simulation Failed] error={:?} signature={:?}", err, signature);
|
||||
warn!(target: "sol_trade_sdk", "[Simulation Failed] error={:?} signature={:?}", err, signature);
|
||||
if let Some(logs) = &simulate_result.value.logs {
|
||||
log::trace!("Transaction logs: {:?}", logs);
|
||||
trace!(target: "sol_trade_sdk", "Transaction logs: {:?}", logs);
|
||||
}
|
||||
if let Some(units_consumed) = simulate_result.value.units_consumed {
|
||||
log::trace!("Compute Units Consumed: {}", units_consumed);
|
||||
trace!(target: "sol_trade_sdk", "Compute Units Consumed: {}", units_consumed);
|
||||
}
|
||||
}
|
||||
return Ok((false, vec![signature], Some(anyhow::anyhow!("{:?}", err))));
|
||||
@@ -270,12 +289,12 @@ async fn simulate_transaction(
|
||||
// Simulation succeeded
|
||||
#[cfg(feature = "perf-trace")]
|
||||
{
|
||||
log::info!("[Simulation Succeeded] signature={:?}", signature);
|
||||
info!(target: "sol_trade_sdk", "[Simulation Succeeded] signature={:?}", signature);
|
||||
if let Some(units_consumed) = simulate_result.value.units_consumed {
|
||||
log::trace!("Compute Units Consumed: {}", units_consumed);
|
||||
trace!(target: "sol_trade_sdk", "Compute Units Consumed: {}", units_consumed);
|
||||
}
|
||||
if let Some(logs) = &simulate_result.value.logs {
|
||||
log::trace!("Transaction logs: {:?}", logs);
|
||||
trace!(target: "sol_trade_sdk", "Transaction logs: {:?}", logs);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -67,6 +67,12 @@ pub struct SwapParams {
|
||||
pub fixed_output_amount: Option<u64>,
|
||||
pub gas_fee_strategy: GasFeeStrategy,
|
||||
pub simulate: bool,
|
||||
/// Whether to output SDK logs (from TradeConfig.log_enabled).
|
||||
pub log_enabled: bool,
|
||||
/// Whether to pin parallel submit tasks to cores (from TradeConfig.use_core_affinity).
|
||||
pub use_core_affinity: bool,
|
||||
/// Optional event receive time in microseconds (same scale as sol-parser-sdk clock::now_micros). Used as timing start when log_enabled.
|
||||
pub grpc_recv_us: Option<i64>,
|
||||
/// Use exact SOL amount instructions (buy_exact_sol_in for PumpFun, buy_exact_quote_in for PumpSwap).
|
||||
/// When Some(true) or None (default), the exact SOL/quote amount is spent and slippage is applied to output tokens.
|
||||
/// When Some(false), uses regular buy instruction where slippage is applied to SOL/quote input.
|
||||
|
||||
Reference in New Issue
Block a user