diff --git a/Cargo.toml b/Cargo.toml index ae79a86..096fd9b 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "sol-trade-sdk" -version = "3.6.3" +version = "3.6.4" edition = "2021" authors = [ "William ", diff --git a/README.md b/README.md index 2a765d7..b732646 100644 --- a/README.md +++ b/README.md @@ -90,14 +90,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.3" } +sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.4" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -sol-trade-sdk = "3.6.3" +sol-trade-sdk = "3.6.4" ``` ## 🛠️ Usage Examples diff --git a/README_CN.md b/README_CN.md index e1e3e21..19365f2 100755 --- a/README_CN.md +++ b/README_CN.md @@ -90,14 +90,14 @@ git clone https://github.com/0xfnzero/sol-trade-sdk ```toml # 添加到您的 Cargo.toml -sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.3" } +sol-trade-sdk = { path = "./sol-trade-sdk", version = "3.6.4" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -sol-trade-sdk = "3.6.3" +sol-trade-sdk = "3.6.4" ``` ## 🛠️ 使用示例 diff --git a/RELEASE_v3.6.3.md b/RELEASE_v3.6.3.md deleted file mode 100644 index 8e4cde5..0000000 --- a/RELEASE_v3.6.3.md +++ /dev/null @@ -1,14 +0,0 @@ -# Release v3.6.3 - -## Changes from v3.6.2 - -### Parallel multi-SWQoS submit - -- **Transaction builder pool**: Introduced `PARALLEL_SENDER_COUNT` (18) and ensured pool prefill is at least 18, so that when using dedicated sender threads all channels can acquire a builder without waiting or allocating. Prefill is now 64 with a guaranteed minimum of 18. -- **Async executor**: Documented that builder pool prefill must match the dedicated sender thread count so multi-channel submit never serializes on `build_transaction`. -- **Executor logging**: Submit timings are no longer sorted before printing (avoids any extra work on the hot path). Log order reflects completion order (first-completed first); the last line is the slowest channel in that batch. - -### Other - -- Added `docs/ASYNC_EXECUTOR_REVIEW.md`. -- Minor updates in types, lib, params, and middleware example. diff --git a/RELEASE_v3.6.4.md b/RELEASE_v3.6.4.md new file mode 100644 index 0000000..7902f0c --- /dev/null +++ b/RELEASE_v3.6.4.md @@ -0,0 +1,8 @@ +# Release v3.6.4 + +## Changes from v3.6.3 + +- **Examples**: All workspace examples are verified to build successfully (`cargo build --workspace`). +- **Docs**: Version references in README.md and README_CN.md updated to 3.6.4. + +No API or behavior changes in this release. diff --git a/docs/ASYNC_EXECUTOR_REVIEW.md b/docs/ASYNC_EXECUTOR_REVIEW.md deleted file mode 100644 index dc2dcef..0000000 --- a/docs/ASYNC_EXECUTOR_REVIEW.md +++ /dev/null @@ -1,24 +0,0 @@ -# async_executor 逻辑与超低延迟 / 无锁竞争 审查 - -## 1. 逻辑正确性 - -- **execute_parallel 流程**：预计算 task_configs → 建一次 shared + collector → 选 queue/notify（专属池或 tokio 池）→ 填 tip_cache、组 SwqosJob、全部 push → notify_waiters → 根据 wait_transaction_confirmed 调 wait_for_success 或 wait_for_all_submitted。逻辑正确。 -- **ResultCollector**：submit 用无锁 ArrayQueue + 原子标志；wait_for_success 先看 success_flag/landed_failed_flag 再 drain results。先成功即返回，未 drain 到的后续结果会随 collector 丢弃，符合「任一成功即返回」语义。 -- **ensure_swqos_pool**：SWQOS_WORKERS_STARTED 用 swap 保证只初始化一次；队列由 execute_parallel 侧 get_or_init，再传给 ensure_swqos_pool，先起 worker 再 push，顺序正确。 -- **ensure_dedicated_pool**：在 Mutex 内判空、创建 queue/notify、spawn 线程、存 JoinHandles，返回 (queue, notify)。首次调用持锁完成初始化，后续调用每次持锁取 (queue, notify) 再返回。 -- **tip_cache**：用 `Arc::as_ptr(&swqos_client)` 做 key，按 client 身份去重，正确。 - -## 2. 锁竞争与超低延迟 - -- **DEDICATED_POOL 的 Mutex**：专属线程池开启时，**每次** execute_parallel 都会调 ensure_dedicated_pool，从而 **每次** 对 DEDICATED_POOL 加锁一次，仅为了读已有的 (queue, notify)。高并发下会成为争用点。 -- **优化**：初始化完成后，queue/notify 存到 OnceCell，热路径只做 OnceCell::get + Arc::clone，不再碰 Mutex。 -- **其余**：ArrayQueue（无锁 MPMC）、ResultCollector（ArrayQueue + 原子变量）、Notify 均无额外锁，合适。 - -## 3. 已实现的优化 - -- **专属池热路径无锁**：`DEDICATED_QUEUE` / `DEDICATED_NOTIFY` 改为 `OnceCell` 存储；`DEDICATED_INIT` 仅存 `JoinHandle` 并在初始化时持锁。热路径先读 OnceCell，命中则直接 `Arc::clone` 返回，不再加锁；未命中时再持锁做一次性初始化并 set OnceCell。 - -## 4. 其他说明 - -- **wait_for_success 与 drain**：先读 `success_flag` 再 `while let Some(...) = results.pop()` 时，若某 worker 刚 store(success) 尚未 push(result)，可能本轮 drain 为空，则 `!signatures.is_empty()` 不成立，不会 return，下一轮轮询会再 drain，逻辑正确。 -- **SWQOS_QUEUE / SWQOS_NOTIFY**：tokio 池侧已用 OnceCell，无每次加锁。 diff --git a/src/common/types.rs b/src/common/types.rs index e793606..6a1c750 100755 --- a/src/common/types.rs +++ b/src/common/types.rs @@ -68,12 +68,6 @@ pub struct TradeConfig { pub create_wsol_ata_on_startup: bool, /// Whether to use seed optimization for all ATA operations (default: true) pub use_seed_optimize: bool, - /// Whether to pin parallel submit tasks to CPU cores (can reduce latency; set false in containers). Default true. - pub use_core_affinity: bool, - /// Use dedicated OS threads for sender pool (opt-in). When true, N threads run only send work; default N=18. Reduces scheduling contention when sending many txs. Default false. - pub use_dedicated_sender_threads: bool, - /// When use_dedicated_sender_threads is true, core indices to pin each sender thread to. If None or empty, N=SWQOS_DEDICATED_DEFAULT_THREADS with no affinity. If Some(ids), N=ids.len() and threads are pinned to these cores. Arc avoids cloning the Vec when building params. - pub sender_thread_cores: Option>>, /// Whether to output all SDK logs (timing, SWQOS submit/confirm, WSOL, blacklist, etc.). Default true. pub log_enabled: bool, /// Whether to check minimum tip per SWQOS provider (filter out configs below min). Default false to save latency. @@ -95,11 +89,8 @@ impl TradeConfig { swqos_configs, commitment, create_wsol_ata_on_startup: true, // default: check and create on startup - use_seed_optimize: true, // default: use seed optimization - use_core_affinity: true, // default: pin parallel submit tasks to cores - use_dedicated_sender_threads: false, // default: use tokio worker pool - sender_thread_cores: None, // when dedicated threads enabled, which cores to pin (None => default count, no affinity) - log_enabled: true, // default: enable all SDK logs + use_seed_optimize: true, // default: use seed optimization + log_enabled: true, // default: enable all SDK logs check_min_tip: false, // default: skip min tip check to reduce latency } } diff --git a/src/lib.rs b/src/lib.rs index 8fb1f46..c3ca99a 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -90,6 +90,10 @@ 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. + 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>, } impl TradingInfrastructure { @@ -175,10 +179,23 @@ impl TradingInfrastructure { } } + let swqos_count = swqos_clients.len(); + 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 ids = core_affinity::get_core_ids() + .map(|all| all.into_iter().take(cap).collect::>()) + .unwrap_or_default(); + (cap, Arc::new(ids)) + }; + Self { rpc, swqos_clients: Arc::new(swqos_clients), config, + max_sender_concurrency, + effective_core_ids, } } } @@ -199,12 +216,14 @@ pub struct TradingClient { /// Whether to use seed optimization for all ATA operations (default: true) /// Applies to all token account creations across buy and sell operations pub use_seed_optimize: bool, - /// Whether to pin parallel submit tasks to CPU cores (from TradeConfig.use_core_affinity). Default true. - pub use_core_affinity: bool, - /// Use dedicated sender threads (from TradeConfig.use_dedicated_sender_threads). Default false. + /// Internal: use dedicated sender threads (default false). Set via with_dedicated_sender_threads() for advanced use. pub use_dedicated_sender_threads: bool, - /// Core indices for dedicated sender threads (from TradeConfig.sender_thread_cores). Arc avoids cloning the Vec when building SwapParams. + /// Internal: core indices for dedicated sender threads. Trimmed to ≤ max_sender_concurrency at set. pub sender_thread_cores: Option>>, + /// Internal: precomputed at infra init (min(swqos_count, 2/3*cores)). Not user-configurable. + pub max_sender_concurrency: usize, + /// Internal: precomputed at infra init for job affinity. Not user-configurable. + pub effective_core_ids: Arc>, /// Whether to output all SDK logs (from TradeConfig.log_enabled). pub log_enabled: bool, /// Whether to check minimum tip per SWQOS (from TradeConfig.check_min_tip). Default false for lower latency. @@ -223,9 +242,10 @@ impl Clone for TradingClient { infrastructure: self.infrastructure.clone(), middleware_manager: self.middleware_manager.clone(), use_seed_optimize: self.use_seed_optimize, - use_core_affinity: self.use_core_affinity, use_dedicated_sender_threads: self.use_dedicated_sender_threads, sender_thread_cores: self.sender_thread_cores.clone(), + max_sender_concurrency: self.max_sender_concurrency, + effective_core_ids: self.effective_core_ids.clone(), log_enabled: self.log_enabled, check_min_tip: self.check_min_tip, } @@ -348,15 +368,18 @@ impl TradingClient { ) -> Self { // Initialize wallet-specific caches (fast, synchronous) crate::common::fast_fn::fast_init(&payer.pubkey()); + let max_sender_concurrency = infrastructure.max_sender_concurrency; + let effective_core_ids = infrastructure.effective_core_ids.clone(); Self { payer, infrastructure, middleware_manager: None, use_seed_optimize, - use_core_affinity: true, use_dedicated_sender_threads: false, sender_thread_cores: None, + max_sender_concurrency, + effective_core_ids, log_enabled: true, check_min_tip: false, } @@ -391,14 +414,18 @@ impl TradingClient { } } + let max_sender_concurrency = infrastructure.max_sender_concurrency; + let effective_core_ids = infrastructure.effective_core_ids.clone(); + Self { payer, infrastructure, middleware_manager: None, use_seed_optimize, - use_core_affinity: true, use_dedicated_sender_threads: false, sender_thread_cores: None, + max_sender_concurrency, + effective_core_ids, log_enabled: true, check_min_tip: false, } @@ -568,14 +595,16 @@ impl TradingClient { } } + // 并发/核心相关由 infrastructure 预计算，用户无需配置 let instance = Self { payer, - infrastructure, + infrastructure: infrastructure.clone(), middleware_manager: None, use_seed_optimize: trade_config.use_seed_optimize, - use_core_affinity: trade_config.use_core_affinity, - use_dedicated_sender_threads: trade_config.use_dedicated_sender_threads, - sender_thread_cores: trade_config.sender_thread_cores.clone(), + use_dedicated_sender_threads: false, + sender_thread_cores: None, + max_sender_concurrency: infrastructure.max_sender_concurrency, + effective_core_ids: infrastructure.effective_core_ids.clone(), log_enabled: trade_config.log_enabled, check_min_tip: trade_config.check_min_tip, }; @@ -601,6 +630,31 @@ impl TradingClient { self } + /// **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). + /// - `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). + pub fn with_dedicated_sender_threads(mut self, core_indices: Option>) -> Self { + match core_indices { + None => { + self.use_dedicated_sender_threads = false; + self.sender_thread_cores = None; + } + Some(v) if v.is_empty() => { + self.use_dedicated_sender_threads = true; + self.sender_thread_cores = None; + } + Some(v) => { + self.use_dedicated_sender_threads = true; + let cap = v.len().min(self.max_sender_concurrency); + self.sender_thread_cores = Some(Arc::new(if cap < v.len() { v[..cap].to_vec() } else { v })); + } + } + self + } + /// Gets the RPC client instance for direct Solana blockchain interactions /// /// This provides access to the underlying Solana RPC client that can be used @@ -721,9 +775,10 @@ impl TradingClient { gas_fee_strategy: params.gas_fee_strategy, simulate: params.simulate, log_enabled: self.log_enabled, - use_core_affinity: self.use_core_affinity, use_dedicated_sender_threads: self.use_dedicated_sender_threads, sender_thread_cores: self.sender_thread_cores.clone(), + max_sender_concurrency: self.max_sender_concurrency, + effective_core_ids: self.effective_core_ids.clone(), check_min_tip: self.check_min_tip, grpc_recv_us: params.grpc_recv_us, use_exact_sol_amount: params.use_exact_sol_amount, @@ -827,9 +882,10 @@ impl TradingClient { gas_fee_strategy: params.gas_fee_strategy, simulate: params.simulate, log_enabled: self.log_enabled, - use_core_affinity: self.use_core_affinity, use_dedicated_sender_threads: self.use_dedicated_sender_threads, sender_thread_cores: self.sender_thread_cores.clone(), + max_sender_concurrency: self.max_sender_concurrency, + effective_core_ids: self.effective_core_ids.clone(), check_min_tip: self.check_min_tip, grpc_recv_us: params.grpc_recv_us, use_exact_sol_amount: None, diff --git a/src/trading/common/transaction_builder.rs b/src/trading/common/transaction_builder.rs index 8ccee7f..179ad0b 100755 --- a/src/trading/common/transaction_builder.rs +++ b/src/trading/common/transaction_builder.rs @@ -25,8 +25,8 @@ fn sol_f64_to_lamports(sol: f64) -> u64 { (lamports.min(u64::MAX as f64)).round() as u64 } -/// Build standard RPC transaction. -/// Takes Arc/context by reference to avoid clone in worker hot path (Arc::clone is cheap but ref is zero-cost). +/// Build standard RPC transaction (worker hot path). +/// Takes Arc/refs only; one Vec allocation (with_capacity), extend_from_slice for business_instructions, no extra clone of payer/rpc/middleware. pub async fn build_transaction( payer: &Arc, _rpc: Option<&Arc>, diff --git a/src/trading/core/async_executor.rs b/src/trading/core/async_executor.rs index 0a25245..46c97cc 100644 --- a/src/trading/core/async_executor.rs +++ b/src/trading/core/async_executor.rs @@ -1,9 +1,10 @@ //! Parallel executor for multi-SWQOS submit. //! +//! **Hot path (submit):** no lock (OnceCell + lock-free ArrayQueue), no `get_core_ids()`, only Arc clones and queue push. //! - **Pool**: Pre-spawned workers (default 18); hot path only enqueues jobs (no per-call tokio::spawn). -//! - **Dedicated threads** (opt-in via TradeConfig): When `use_dedicated_sender_threads` is true, N OS threads (default 18) run sender work only, optionally pinned to cores via `sender_thread_cores`, reducing scheduling contention when sending many txs. -//! - **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). +//! - **Dedicated threads** (opt-in via `with_dedicated_sender_threads`): N OS threads run sender work only, optionally pinned to cores. +//! - **Arc**: Shared data behind `Arc` → clone = refcount increment (no data copy). +//! - **Refs**: `build_transaction` takes refs only; worker path avoids extra clones. use anyhow::{anyhow, Result}; use crossbeam_queue::ArrayQueue; @@ -15,8 +16,8 @@ use solana_sdk::{ }; use std::collections::HashMap; use std::hash::BuildHasherDefault; +use parking_lot::Mutex; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::sync::Mutex; use std::{str::FromStr, sync::Arc, time::Instant}; use tokio::sync::Notify; @@ -28,6 +29,7 @@ use crate::{ common::nonce_cache::DurableNonceInfo, common::{GasFeeStrategy, SolanaRpcClient}, swqos::{SwqosClient, SwqosType, TradeType}, + trading::core::params::SenderConcurrencyConfig, trading::{common::build_transaction, MiddlewareManager}, }; @@ -154,28 +156,35 @@ static DEDICATED_NOTIFY: OnceCell> = OnceCell::new(); /// 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(sender_thread_cores: Option<&[usize]>) -> (Arc>, Arc) { +fn ensure_dedicated_pool( + 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().expect("dedicated init mutex"); + 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 - .map(|v| v.len()) - .unwrap_or(SWQOS_DEDICATED_DEFAULT_THREADS) - .min(32); + .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 = sender_thread_cores - .and_then(|indices| { - core_affinity::get_core_ids().map(|ids| { - indices - .iter() - .filter_map(|&i| ids.get(i).cloned()) - .collect() - }) + let core_ids: Vec = core_affinity::get_core_ids() + .map(|all_ids| { + sender_thread_cores + .map(|indices| { + indices + .iter() + .take(n) + .filter_map(|&i| all_ids.get(i).cloned()) + .collect() + }) + .unwrap_or_else(|| all_ids.into_iter().take(n).collect()) }) .unwrap_or_default(); let mut handles = Vec::with_capacity(n); @@ -201,12 +210,13 @@ fn ensure_dedicated_pool(sender_thread_cores: Option<&[usize]>) -> (Arc>) { +fn ensure_swqos_pool(queue: Arc>, max_sender_concurrency: usize) { if SWQOS_WORKERS_STARTED.swap(true, Ordering::AcqRel) { 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..SWQOS_POOL_WORKERS { + for _ in 0..n { tokio::spawn(swqos_worker_loop(queue.clone(), notify.clone())); } } @@ -400,6 +410,8 @@ impl ResultCollector { } /// 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). pub async fn execute_parallel( swqos_clients: &[Arc], payer: Arc, @@ -414,13 +426,10 @@ pub async fn execute_parallel( wait_transaction_confirmed: bool, with_tip: bool, gas_fee_strategy: GasFeeStrategy, - use_core_affinity: bool, use_dedicated_sender_threads: bool, - sender_thread_cores: Option<&[usize]>, + sender_config: SenderConcurrencyConfig, check_min_tip: bool, ) -> Result<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { - let _exec_start = Instant::now(); - if swqos_clients.is_empty() { return Err(anyhow!("swqos_clients is empty")); } @@ -434,44 +443,40 @@ pub async fn execute_parallel( return Err(anyhow!("No Rpc Default Swqos configured.")); } - 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() - .filter(|(_, swqos_client)| { - with_tip || matches!(swqos_client.get_swqos_type(), SwqosType::Default) - }) - .flat_map(|(i, swqos_client)| { - let swqos_type = swqos_client.get_swqos_type(); - let gas_fee_strategy_configs = gas_fee_strategy.get_strategies(if is_buy { - TradeType::Buy - } else { - TradeType::Sell - }); - 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 }; - gas_fee_strategy_configs - .into_iter() - .filter(move |config| config.0 == swqos_type) - .filter(move |config| { - 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 - ); - } - config.2.tip >= min_tip - } else { - true - } - }) - .map(move |config| (i, swqos_client.clone(), config)) - }) - .collect(); + // 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() { + if !with_tip && !matches!(swqos_client.get_swqos_type(), SwqosType::Default) { + continue; + } + let swqos_type = swqos_client.get_swqos_type(); + 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)); + } + } if task_configs.is_empty() { return Err(anyhow!("No available gas fee strategy configs")); @@ -499,23 +504,27 @@ pub async fn execute_parallel( }); let (queue, notify) = if use_dedicated_sender_threads { - ensure_dedicated_pool(sender_thread_cores) + ensure_dedicated_pool( + sender_config.sender_thread_cores.as_ref().map(|a| a.as_slice()), + sender_config.max_sender_concurrency, + ) } else { let q = SWQOS_QUEUE.get_or_init(|| Arc::new(ArrayQueue::new(SWQOS_QUEUE_CAP))); - ensure_swqos_pool(q.clone()); + ensure_swqos_pool(q.clone(), sender_config.max_sender_concurrency); (q.clone(), SWQOS_NOTIFY.get_or_init(|| Arc::new(Notify::new())).clone()) }; { - // Cache tip_account per client (one get_tip_account/from_str per unique client per batch). Dropped before await so future stays Send. + 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 = cores.get(i % cores.len().max(1)).copied(); + let core_id = effective_core_ids.get(i % core_len).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(), + Some(t) => t.clone(), None => { let s = swqos_client.get_tip_account()?; let tip = Arc::new(Pubkey::from_str(&s).unwrap_or_default()); @@ -537,7 +546,7 @@ pub async fn execute_parallel( swqos_client, swqos_type, core_id, - use_affinity: use_core_affinity, + use_affinity: !effective_core_ids.is_empty(), }; let _ = queue.push(job); } diff --git a/src/trading/core/executor.rs b/src/trading/core/executor.rs index cc694de..3864617 100755 --- a/src/trading/core/executor.rs +++ b/src/trading/core/executor.rs @@ -144,6 +144,7 @@ impl TradeExecutor for GenericTradeExecutor { } let need_confirm = params.wait_transaction_confirmed; + let sender_config = params.sender_concurrency_config(); let result = execute_parallel( params.swqos_clients.as_slice(), params.payer, @@ -158,9 +159,8 @@ impl TradeExecutor for GenericTradeExecutor { 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, params.use_dedicated_sender_threads, - params.sender_thread_cores.as_ref().map(|a| a.as_slice()), + sender_config, params.check_min_tip, ) .await; diff --git a/src/trading/core/params.rs b/src/trading/core/params.rs index 5e27daa..0701e0e 100755 --- a/src/trading/core/params.rs +++ b/src/trading/core/params.rs @@ -3,7 +3,16 @@ use crate::common::nonce_cache::DurableNonceInfo; use crate::common::spl_associated_token_account::get_associated_token_address_with_program_id; use crate::common::{GasFeeStrategy, SolanaRpcClient}; use crate::constants::TOKEN_PROGRAM; +use core_affinity::CoreId; use crate::instruction::utils::pumpfun::global_constants::MAYHEM_FEE_RECIPIENT; + +/// Concurrency + core binding config for parallel submit (precomputed at SDK init, one param on hot path). Uses Arc so no borrow of SwapParams. +#[derive(Clone)] +pub struct SenderConcurrencyConfig { + pub sender_thread_cores: Option>>, + pub effective_core_ids: Arc>, + pub max_sender_concurrency: usize, +} use crate::instruction::utils::pumpswap::accounts::MAYHEM_FEE_RECIPIENT as MAYHEM_FEE_RECIPIENT_SWAP; use crate::swqos::{SwqosClient, TradeType}; use crate::trading::common::get_multi_token_balances; @@ -70,12 +79,14 @@ pub struct SwapParams { 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, - /// Use dedicated sender threads (from TradeConfig.use_dedicated_sender_threads). + /// Use dedicated sender threads (internal; set via client.with_dedicated_sender_threads()). pub use_dedicated_sender_threads: bool, /// Core indices for dedicated sender threads (from TradeConfig.sender_thread_cores). Arc avoids cloning the Vec on hot path. pub sender_thread_cores: Option>>, + /// Precomputed at SDK init: min(swqos_count, 2/3*cores). Avoids get_core_ids() on trade hot path. + pub max_sender_concurrency: usize, + /// Precomputed at SDK init: first max_sender_concurrency CoreIds for job affinity. Arc clone only. + pub effective_core_ids: Arc>, /// Whether to check minimum tip per SWQOS (from TradeConfig.check_min_tip). When false, skip filter for lower latency. pub check_min_tip: bool, /// Optional event receive time in microseconds (same scale as sol-parser-sdk clock::now_micros). Used as timing start when log_enabled. @@ -87,6 +98,18 @@ pub struct SwapParams { pub use_exact_sol_amount: Option, } +impl SwapParams { + /// One struct for execute_parallel: merges sender_thread_cores, effective_core_ids, max_sender_concurrency. Arc clone only. + #[inline] + pub fn sender_concurrency_config(&self) -> SenderConcurrencyConfig { + SenderConcurrencyConfig { + sender_thread_cores: self.sender_thread_cores.clone(), + effective_core_ids: self.effective_core_ids.clone(), + max_sender_concurrency: self.max_sender_concurrency, + } + } +} + impl std::fmt::Debug for SwapParams { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "SwapParams: ...")