gavindiaz/sol-trade-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: P0/P1/P2 low-latency and maintainability improvements

Browse Source 
P0:
- blockhash/nonce: get_transaction_blockhash returns Result; buy/sell entry validation
- Bonk sell: remove RPC get_token_balance from build path; require caller to pass input_amount
- Hot path: drop redundant dex_type/protocol_params clone; tip via sol_f64_to_lamports (no String)
- compute_budget cache uses Arc; extend_compute_budget_instructions avoids SmallVec clone

P1:
- middleware protocol_name takes &str; executor destructures result once to avoid signatures/submit_timings clone
- Error messages include dex context; params use Copy for Pubkey; pumpswap utils clone only Pool for pools[0]

P2:
- transaction_pool capacity constants; TIP_ACCOUNT_CACHE marked allow(dead_code)
- Fix error copy in raydium_amm_v4 / meteora_damm_v2 to correct protocol names

Made-with: Cursor
This commit is contained in:
Wood
2026-03-08 01:05:27 +08:00
parent e32e7ee6ab
commit 9f79e865ab
17 changed files with 397 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trading/core/async_executor.rs
+187 -101
View File
@@ -1,15 +1,26 @@
//! Parallel executor for multi-SWQOS submit.
//!
//! - **Pool**: Pre-spawned workers; hot path only enqueues jobs (no per-call tokio::spawn).
//! - **Arc**: Shared data is behind `Arc` so "clone" is just a refcount increment (no data copy).
//! - **Refs**: `build_transaction` takes `&Arc<..>`, `Option<&DurableNonceInfo>`, `Option<&AddressLookupTableAccount>` so the worker passes refs only (zero clone on worker path).
use anyhow::{anyhow, Result};
use crossbeam_queue::ArrayQueue;
use once_cell::sync::OnceCell;
use solana_hash::Hash;
use solana_sdk::message::AddressLookupTableAccount;
use solana_sdk::{
instruction::Instruction, pubkey::Pubkey, signature::Keypair, signature::Signature,
};
use std::collections::HashMap;
use std::hash::BuildHasherDefault;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::{str::FromStr, sync::Arc, time::Instant};
use fnv::FnvHasher;
type FnvHashMap<K, V> = HashMap<K, V, BuildHasherDefault<FnvHasher>>;
use crate::{
common::nonce_cache::DurableNonceInfo,
common::{GasFeeStrategy, SolanaRpcClient},
@@ -17,6 +28,133 @@ use crate::{
trading::{common::build_transaction, MiddlewareManager},
};
const SWQOS_POOL_WORKERS: usize = 32;
const SWQOS_QUEUE_CAP: usize = 128;
/// Shared across all jobs in one batch; built once, cloned as single Arc per job (minimal hot-path clone).
struct SwqosSharedContext {
payer: Arc<Keypair>,
instructions: Arc<Vec<Instruction>>,
rpc: Option<Arc<SolanaRpcClient>>,
address_lookup_table_account: Option<AddressLookupTableAccount>,
recent_blockhash: Option<Hash>,
durable_nonce: Option<DurableNonceInfo>,
middleware_manager: Option<Arc<MiddlewareManager>>,
protocol_name: &'static str,
is_buy: bool,
wait_transaction_confirmed: bool,
with_tip: bool,
collector: Arc<ResultCollector>,
}
/// One SWQOS submit task; only per-task data + one Arc to shared (reduces hot-path clones).
struct SwqosJob {
shared: Arc<SwqosSharedContext>,
tip: f64,
unit_limit: u32,
unit_price: u64,
tip_account: Arc<Pubkey>,
swqos_client: Arc<SwqosClient>,
swqos_type: SwqosType,
core_id: Option<core_affinity::CoreId>,
use_affinity: bool,
}
async fn run_one_swqos_job(job: SwqosJob) {
let s = &job.shared;
if job.use_affinity {
if let Some(cid) = job.core_id {
core_affinity::set_for_current(cid);
}
}
let tip_amount = if s.with_tip { job.tip } else { 0.0 };
let transaction = match build_transaction(
&s.payer,
s.rpc.as_ref(),
job.unit_limit,
job.unit_price,
s.instructions.as_ref(),
s.address_lookup_table_account.as_ref(),
s.recent_blockhash,
s.middleware_manager.as_ref(),
s.protocol_name,
s.is_buy,
job.swqos_type != SwqosType::Default,
&job.tip_account,
tip_amount,
s.durable_nonce.as_ref(),
)
.await
{
Ok(tx) => tx,
Err(e) => {
s.collector.submit(TaskResult {
success: false,
signature: Signature::default(),
error: Some(e),
swqos_type: job.swqos_type,
landed_on_chain: false,
submit_done_us: crate::common::clock::now_micros(),
});
return;
}
};
let (success, err, landed_on_chain) = match job
.swqos_client
.send_transaction(
if s.is_buy {
TradeType::Buy
} else {
TradeType::Sell
},
&transaction,
s.wait_transaction_confirmed,
)
.await
{
Ok(()) => (true, None, true),
Err(e) => {
let landed = is_landed_error(&e);
(false, Some(e), landed)
}
};
let sig = transaction.signatures.first().copied().unwrap_or_default();
s.collector.submit(TaskResult {
success,
signature: sig,
error: err,
swqos_type: job.swqos_type,
landed_on_chain,
submit_done_us: crate::common::clock::now_micros(),
});
}
async fn swqos_worker_loop(queue: Arc<ArrayQueue<SwqosJob>>) {
loop {
if let Some(job) = queue.pop() {
run_one_swqos_job(job).await;
} else {
tokio::task::yield_now().await;
}
}
}
static SWQOS_QUEUE: OnceCell<Arc<ArrayQueue<SwqosJob>>> = OnceCell::new();
static SWQOS_WORKERS_STARTED: AtomicBool = AtomicBool::new(false);
fn ensure_swqos_pool(queue: Arc<ArrayQueue<SwqosJob>>) {
if SWQOS_WORKERS_STARTED.swap(true, Ordering::AcqRel) {
return;
}
for _ in 0..SWQOS_POOL_WORKERS {
tokio::spawn(swqos_worker_loop(queue.clone()));
}
}
#[repr(align(64))]
struct TaskResult {
success: bool,
@@ -282,116 +420,64 @@ pub async fn execute_parallel(
return Err(anyhow!("Multiple swqos transactions require durable_nonce to be set.",));
}
// Task preparation completed
// Task preparation completed: one shared context (clone once per batch), then minimal per-task data.
let collector = Arc::new(ResultCollector::new(task_configs.len()));
let _spawn_start = Instant::now();
let shared = Arc::new(SwqosSharedContext {
payer,
instructions,
rpc,
address_lookup_table_account,
recent_blockhash,
durable_nonce,
middleware_manager,
protocol_name,
is_buy,
wait_transaction_confirmed,
with_tip,
collector: collector.clone(),
});
for (i, swqos_client, gas_fee_strategy_config) in task_configs {
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();
let swqos_type = swqos_client.get_swqos_type();
let tip_account_str = swqos_client.get_tip_account()?;
let tip_account = Arc::new(Pubkey::from_str(&tip_account_str).unwrap_or_default());
let collector = collector.clone();
let queue = SWQOS_QUEUE.get_or_init(|| Arc::new(ArrayQueue::new(SWQOS_QUEUE_CAP)));
ensure_swqos_pool(queue.clone());
let tip = gas_fee_strategy_config.2.tip;
let unit_limit = gas_fee_strategy_config.2.cu_limit;
let unit_price = gas_fee_strategy_config.2.cu_price;
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();
if use_affinity {
if let Some(cid) = core_id {
core_affinity::set_for_current(cid);
{
// Cache tip_account per client (one get_tip_account/from_str per unique client per batch). Dropped before await so future stays Send.
let mut tip_cache: FnvHashMap<*const (), Arc<Pubkey>> =
FnvHashMap::with_capacity_and_hasher(task_configs.len(), BuildHasherDefault::default());
for (i, swqos_client, gas_fee_strategy_config) in task_configs {
let core_id = cores.get(i % cores.len().max(1)).copied();
let swqos_type = swqos_client.get_swqos_type();
let key = Arc::as_ptr(&swqos_client) as *const ();
let tip_account = match tip_cache.get(&key) {
Some(tip) => tip.clone(),
None => {
let s = swqos_client.get_tip_account()?;
let tip = Arc::new(Pubkey::from_str(&s).unwrap_or_default());
tip_cache.insert(key, tip.clone());
tip
}
}
let tip_amount = if with_tip { tip } else { 0.0 };
let _build_start = Instant::now();
let transaction = match build_transaction(
payer,
rpc,
};
let (tip, unit_limit, unit_price) = (
gas_fee_strategy_config.2.tip,
gas_fee_strategy_config.2.cu_limit,
gas_fee_strategy_config.2.cu_price,
);
let job = SwqosJob {
shared: shared.clone(),
tip,
unit_limit,
unit_price,
instructions.as_ref(),
address_lookup_table_account,
recent_blockhash_task,
middleware_manager,
protocol_name,
is_buy,
swqos_type != SwqosType::Default,
&tip_account,
tip_amount,
durable_nonce,
)
.await
{
Ok(tx) => tx,
Err(e) => {
// Build transaction failed
collector.submit(TaskResult {
success: false,
signature: Signature::default(),
error: Some(e),
swqos_type,
landed_on_chain: false,
submit_done_us: crate::common::clock::now_micros(),
});
return;
}
};
// Transaction built
let _send_start = Instant::now();
let mut err: Option<anyhow::Error> = None;
#[allow(unused_assignments)]
let mut landed_on_chain = false;
let success = match swqos_client
.send_transaction(
if is_buy { TradeType::Buy } else { TradeType::Sell },
&transaction,
wait_transaction_confirmed,
)
.await
{
Ok(()) => {
landed_on_chain = true; // Success means tx confirmed on-chain
true
}
Err(e) => {
// Check if this error indicates the tx landed but failed (e.g., ExceededSlippage)
landed_on_chain = is_landed_error(&e);
err = Some(e);
// Send transaction failed
false
}
};
// 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,
tip_account,
swqos_client,
swqos_type,
landed_on_chain,
submit_done_us: crate::common::clock::now_micros(),
});
});
core_id,
use_affinity: use_core_affinity,
};
let _ = queue.push(job);
}
}
// All tasks spawned
// All jobs enqueued (no spawn on hot path)
if !wait_transaction_confirmed {
const SUBMIT_TIMEOUT_SECS: u64 = 30;
src/trading/core/executor.rs
+44 -47
View File
@@ -77,7 +77,7 @@ impl TradeExecutor for GenericTradeExecutor {
Some(middleware_manager) => middleware_manager
.apply_middlewares_process_protocol_instructions(
instructions,
self.protocol_name.to_string(),
self.protocol_name,
is_buy,
)?,
None => instructions,
@@ -144,53 +144,51 @@ impl TradeExecutor for GenericTradeExecutor {
.await;
let log_enabled = params.log_enabled && crate::common::sdk_log::sdk_log_enabled();
let submit_timings = if log_enabled {
result.as_ref().ok().map(|(_, _, _, t)| t.clone()).unwrap_or_default()
} else {
Vec::new()
let (ok, signatures, err, submit_timings) = match result {
Ok((success, sigs, last_error, timings)) => (
success,
sigs,
last_error.map(|e| anyhow::anyhow!("{}", e)),
timings,
),
Err(e) => (false, vec![], Some(anyhow::anyhow!("{}", e)), vec![]),
};
let submit_timings_ref: &[(crate::swqos::SwqosType, i64)] = submit_timings.as_slice();
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 confirm_result = if let Some(rpc) = params.rpc.as_ref() {
if sigs.is_empty() {
(ok, sigs, err)
if signatures.is_empty() {
(ok, signatures, err)
} else {
let poll_res = poll_any_transaction_confirmation(rpc, &sigs, true).await;
let confirm_done_us = log_enabled.then(crate::common::clock::now_micros);
if log_enabled {
let dir = if is_buy { "Buy" } else { "Sell" };
if let Some(start_us) = timing_start_us {
if let Some(end_us) = build_end_us {
println!(" [SDK] {} build_instructions: {:.4} ms", dir, (end_us - start_us) as f64 / 1000.0);
}
if let Some(end_us) = before_submit_us {
println!(" [SDK] {} before_submit: {:.4} ms", dir, (end_us - start_us) as f64 / 1000.0);
}
if let Some(confirm_us) = confirm_done_us {
let total_ms = (confirm_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_us - *submit_done_us).max(0) as f64 / 1000.0;
println!(" [SDK] {} {:?} submit: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", dir, swqos_type, submit_ms, confirmed_ms, total_ms);
let poll_res = poll_any_transaction_confirmation(rpc, &signatures, true).await;
let confirm_done_us = log_enabled.then(crate::common::clock::now_micros);
if log_enabled {
let dir = if is_buy { "Buy" } else { "Sell" };
if let Some(start_us) = timing_start_us {
if let Some(end_us) = build_end_us {
println!(" [SDK] {} build_instructions: {:.4} ms", dir, (end_us - start_us) as f64 / 1000.0);
}
if let Some(end_us) = before_submit_us {
println!(" [SDK] {} before_submit: {:.4} ms", dir, (end_us - start_us) as f64 / 1000.0);
}
if let Some(confirm_us) = confirm_done_us {
let total_ms = (confirm_us - start_us) as f64 / 1000.0;
for (swqos_type, submit_done_us) in submit_timings_ref {
let submit_ms = (*submit_done_us - start_us).max(0) as f64 / 1000.0;
let confirmed_ms = (confirm_us - *submit_done_us).max(0) as f64 / 1000.0;
println!(" [SDK] {} {:?} submit: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", dir, swqos_type, submit_ms, confirmed_ms, total_ms);
}
}
}
}
}
match poll_res {
Ok(_) => (true, sigs, None),
Err(e) => (false, sigs, Some(e)),
}
match poll_res {
Ok(_) => (true, signatures, None),
Err(e) => (false, signatures, Some(e)),
}
}
} else {
(ok, sigs, err)
(ok, signatures, err)
};
Ok(confirm_result)
} else {
@@ -203,13 +201,13 @@ impl TradeExecutor for GenericTradeExecutor {
if let Some(end_us) = before_submit_us {
println!(" [SDK] {} before_submit: {:.4} ms", dir, (end_us - start_us) as f64 / 1000.0);
}
for (swqos_type, submit_done_us) in &submit_timings {
for (swqos_type, submit_done_us) in submit_timings_ref {
let submit_ms = (*submit_done_us - start_us).max(0) as f64 / 1000.0;
println!(" [SDK] {} {:?} submit: {:.4} ms, confirmed: -, total: {:.4} ms", dir, swqos_type, submit_ms, submit_ms);
}
}
}
result.map(|(a, b, c, _)| (a, b, c))
Ok((ok, signatures, err))
};
result
@@ -256,22 +254,21 @@ async fn simulate_transaction(
let unit_limit = default_config.2.cu_limit;
let unit_price = default_config.2.cu_price;
// Build transaction for simulation
let transaction = build_transaction(
payer.clone(),
Some(rpc.clone()),
&payer,
Some(&rpc),
unit_limit,
unit_price,
&instructions,
address_lookup_table_account,
address_lookup_table_account.as_ref(),
recent_blockhash,
middleware_manager,
middleware_manager.as_ref(),
protocol_name,
is_buy,
false, // simulate doesn't need tip instruction
false,
&Pubkey::default(),
tip,
durable_nonce,
durable_nonce.as_ref(),
)
.await?;
src/trading/core/params.rs
+3 -3
View File
@@ -398,7 +398,7 @@ impl PumpSwapParams {
);
Ok(Self {
pool: pool_address.clone(),
pool: *pool_address,
base_mint: pool_data.base_mint,
quote_mint: pool_data.quote_mint,
pool_base_token_account: pool_data.pool_base_token_account,
@@ -671,7 +671,7 @@ impl RaydiumCpmmParams {
)
.await?;
Ok(Self {
pool_state: pool_address.clone(),
pool_state: *pool_address,
amm_config: pool.amm_config,
base_mint: pool.token0_mint,
quote_mint: pool.token1_mint,
@@ -778,7 +778,7 @@ impl MeteoraDammV2Params {
let pool_data =
crate::instruction::utils::meteora_damm_v2::fetch_pool(rpc, pool_address).await?;
Ok(Self {
pool: pool_address.clone(),
pool: *pool_address,
token_a_vault: pool_data.token_a_vault,
token_b_vault: pool_data.token_b_vault,
token_a_mint: pool_data.token_a_mint,
src/trading/core/transaction_pool.rs
+19 -14
View File
@@ -6,6 +6,15 @@
//! - 零拷贝 I/O
//! - 内存预热
/// 预分配指令容量（单笔交易常见指令数）
const TX_BUILDER_INSTRUCTION_CAP: usize = 32;
/// 预分配地址查找表数量
const TX_BUILDER_LOOKUP_TABLE_CAP: usize = 8;
/// 对象池最大容量
const TX_BUILDER_POOL_CAP: usize = 1000;
/// 启动时预填充对象池数量
const TX_BUILDER_POOL_PREFILL: usize = 100;
use crossbeam_queue::ArrayQueue;
use once_cell::sync::Lazy;
use solana_sdk::{
@@ -23,8 +32,8 @@ pub struct PreallocatedTxBuilder {
impl PreallocatedTxBuilder {
fn new() -> Self {
Self {
instructions: Vec::with_capacity(32), // 预分配32条指令空间
lookup_tables: Vec::with_capacity(8), // 预分配8个查找表空间
instructions: Vec::with_capacity(TX_BUILDER_INSTRUCTION_CAP),
lookup_tables: Vec::with_capacity(TX_BUILDER_LOOKUP_TABLE_CAP),
}
}
@@ -65,23 +74,20 @@ impl PreallocatedTxBuilder {
&mut self,
payer: &Pubkey,
instructions: &[Instruction],
address_lookup_table_account: Option<AddressLookupTableAccount>,
address_lookup_table_account: Option<&AddressLookupTableAccount>,
recent_blockhash: Hash,
) -> VersionedMessage {
// 重用已分配的 vector
self.reset();
self.instructions.extend_from_slice(instructions);
// ✅ 如果有查找表，使用 V0 消息
if let Some(address_lookup_table_account) = address_lookup_table_account {
let message = v0::Message::try_compile(
if let Some(alt) = address_lookup_table_account {
let message = v0::Message::try_compile(
payer,
&self.instructions,
&[address_lookup_table_account],
std::slice::from_ref(alt),
recent_blockhash,
).expect("v0 message compile failed");
)
.expect("v0 message compile failed");
VersionedMessage::V0(message)
} else {
// ✅ 没有查找表，使用 Legacy 消息（兼容所有 RPC）
@@ -97,10 +103,9 @@ impl PreallocatedTxBuilder {
/// 🚀 全局交易构建器对象池
static TX_BUILDER_POOL: Lazy<Arc<ArrayQueue<PreallocatedTxBuilder>>> = Lazy::new(|| {
let pool = ArrayQueue::new(1000); // 1000个预分配构建器
let pool = ArrayQueue::new(TX_BUILDER_POOL_CAP);
// 预填充池
for _ in 0..100 {
for _ in 0..TX_BUILDER_POOL_PREFILL {
let _ = pool.push(PreallocatedTxBuilder::new());
}