From e957bc4bee04c9f8bd9e9d53596295464b610dac Mon Sep 17 00:00:00 2001 From: Wood Date: Tue, 17 Mar 2026 03:32:59 +0800 Subject: [PATCH] Parallel multi-SWQoS submit: builder pool and executor tweaks - transaction_pool: add PARALLEL_SENDER_COUNT (18), ensure prefill >= 18 so multi-channel build never serializes; prefill 64 with max(PREFILL, 18) - async_executor: document that pool prefill must match sender thread count - executor: do not sort submit_timings (avoids any extra work); comment that log order is completion order - docs: add ASYNC_EXECUTOR_REVIEW.md - remove RELEASE_NOTES_v3.6.2.md; minor example/types/lib/params updates Made-with: Cursor --- RELEASE_NOTES_v3.6.2.md | 16 ----- docs/ASYNC_EXECUTOR_REVIEW.md | 24 ++++++++ examples/middleware_system/src/main.rs | 4 +- src/common/types.rs | 6 ++ src/lib.rs | 26 ++++++++- src/trading/core/async_executor.rs | 81 ++++++++++++++++++++++---- src/trading/core/executor.rs | 7 ++- src/trading/core/params.rs | 7 ++- src/trading/core/transaction_pool.rs | 11 ++-- 9 files changed, 143 insertions(+), 39 deletions(-) delete mode 100644 RELEASE_NOTES_v3.6.2.md create mode 100644 docs/ASYNC_EXECUTOR_REVIEW.md diff --git a/RELEASE_NOTES_v3.6.2.md b/RELEASE_NOTES_v3.6.2.md deleted file mode 100644 index ae2c767..0000000 --- a/RELEASE_NOTES_v3.6.2.md +++ /dev/null @@ -1,16 +0,0 @@ -## sol-trade-sdk v3.6.2 - -### Changes - -- **Node1 QUIC support** — Node1 SWQOS can use QUIC transport: `SwqosConfig::Node1(api_token, region, custom_url, Some(SwqosTransport::Quic))`. Uses UUID auth on first bi stream, one bi stream per transaction, bincode-serialized `VersionedTransaction`. Region endpoints: `SWQOS_ENDPOINTS_NODE1_QUIC` (ny, fra, ams, lon, tk). New dependency: `uuid`. -- **Speedlanding QUIC reliability** — Proactive connection check before send (`ensure_connected()`); 5s connect and send timeouts; reconnect uses `lock().await` so concurrent senders wait for the new connection instead of failing on `try_lock()`; TLS SNI is derived from the endpoint host (e.g. `nyc.speedlanding.trade`) with fallback to `speed-landing` for IP or unknown host. Addresses user reports of transactions failing to send. - -### Crates.io - -```toml -sol-trade-sdk = "3.6.2" -``` - -### Repository - -- **Tag:** [v3.6.2](https://github.com/0xfnzero/sol-trade-sdk/releases/tag/v3.6.2) diff --git a/docs/ASYNC_EXECUTOR_REVIEW.md b/docs/ASYNC_EXECUTOR_REVIEW.md new file mode 100644 index 0000000..dc2dcef --- /dev/null +++ b/docs/ASYNC_EXECUTOR_REVIEW.md @@ -0,0 +1,24 @@ +# 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/examples/middleware_system/src/main.rs b/examples/middleware_system/src/main.rs index e66a4c3..40ebf5b 100644 --- a/examples/middleware_system/src/main.rs +++ b/examples/middleware_system/src/main.rs @@ -31,7 +31,7 @@ impl InstructionMiddleware for CustomMiddleware { fn process_protocol_instructions( &self, protocol_instructions: Vec, - _protocol_name: String, + _protocol_name: &str, _is_buy: bool, ) -> Result> { // do anything you want here @@ -42,7 +42,7 @@ impl InstructionMiddleware for CustomMiddleware { fn process_full_instructions( &self, full_instructions: Vec, - _protocol_name: String, + _protocol_name: &str, _is_buy: bool, ) -> Result> { // do anything you want here diff --git a/src/common/types.rs b/src/common/types.rs index f50fd99..e793606 100755 --- a/src/common/types.rs +++ b/src/common/types.rs @@ -70,6 +70,10 @@ pub struct TradeConfig { 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. @@ -93,6 +97,8 @@ impl TradeConfig { 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 check_min_tip: false, // default: skip min tip check to reduce latency } diff --git a/src/lib.rs b/src/lib.rs index 90ad258..8fb1f46 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -86,8 +86,8 @@ pub enum TradeTokenType { pub struct TradingInfrastructure { /// Shared RPC client for blockchain interactions pub rpc: Arc, - /// Shared SWQOS clients for transaction priority and routing - pub swqos_clients: Vec>, + /// Shared SWQOS clients for transaction priority and routing. Arc> so cloning into SwapParams is a single Arc clone. + pub swqos_clients: Arc>>, /// Configuration used to create this infrastructure pub config: InfrastructureConfig, } @@ -175,7 +175,11 @@ impl TradingInfrastructure { } } - Self { rpc, swqos_clients, config } + Self { + rpc, + swqos_clients: Arc::new(swqos_clients), + config, + } } } @@ -197,6 +201,10 @@ pub struct TradingClient { 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. + 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. + pub sender_thread_cores: Option>>, /// 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. @@ -216,6 +224,8 @@ impl Clone for TradingClient { 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(), log_enabled: self.log_enabled, check_min_tip: self.check_min_tip, } @@ -345,6 +355,8 @@ impl TradingClient { middleware_manager: None, use_seed_optimize, use_core_affinity: true, + use_dedicated_sender_threads: false, + sender_thread_cores: None, log_enabled: true, check_min_tip: false, } @@ -385,6 +397,8 @@ impl TradingClient { middleware_manager: None, use_seed_optimize, use_core_affinity: true, + use_dedicated_sender_threads: false, + sender_thread_cores: None, log_enabled: true, check_min_tip: false, } @@ -560,6 +574,8 @@ impl TradingClient { 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(), log_enabled: trade_config.log_enabled, check_min_tip: trade_config.check_min_tip, }; @@ -706,6 +722,8 @@ impl TradingClient { 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(), check_min_tip: self.check_min_tip, grpc_recv_us: params.grpc_recv_us, use_exact_sol_amount: params.use_exact_sol_amount, @@ -810,6 +828,8 @@ impl TradingClient { 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(), check_min_tip: self.check_min_tip, grpc_recv_us: params.grpc_recv_us, use_exact_sol_amount: None, diff --git a/src/trading/core/async_executor.rs b/src/trading/core/async_executor.rs index 99acf6f..0a25245 100644 --- a/src/trading/core/async_executor.rs +++ b/src/trading/core/async_executor.rs @@ -1,6 +1,7 @@ //! Parallel executor for multi-SWQOS submit. //! -//! - **Pool**: Pre-spawned workers; hot path only enqueues jobs (no per-call tokio::spawn). +//! - **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). @@ -15,6 +16,7 @@ use solana_sdk::{ use std::collections::HashMap; use std::hash::BuildHasherDefault; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::Mutex; use std::{str::FromStr, sync::Arc, time::Instant}; use tokio::sync::Notify; @@ -29,8 +31,10 @@ use crate::{ trading::{common::build_transaction, MiddlewareManager}, }; -const SWQOS_POOL_WORKERS: usize = 32; +/// 与 transaction_pool::PARALLEL_SENDER_COUNT 一致，保证多路 build 不串行 +const SWQOS_POOL_WORKERS: usize = 18; const SWQOS_QUEUE_CAP: usize = 128; +const SWQOS_DEDICATED_DEFAULT_THREADS: usize = 18; /// Shared across all jobs in one batch; built once, cloned as single Arc per job (minimal hot-path clone). struct SwqosSharedContext { @@ -144,6 +148,59 @@ static SWQOS_QUEUE: OnceCell>> = OnceCell::new(); static SWQOS_NOTIFY: OnceCell> = OnceCell::new(); static SWQOS_WORKERS_STARTED: AtomicBool = AtomicBool::new(false); +/// 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(); +/// 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) { + 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"); + 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); + 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() + }) + }) + .unwrap_or_default(); + let mut handles = Vec::with_capacity(n); + for i in 0..n { + let queue = queue.clone(); + let notify = notify.clone(); + let core_id = core_ids.get(i).cloned(); + let handle = std::thread::spawn(move || { + if let Some(cid) = core_id { + core_affinity::set_for_current(cid); + } + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("dedicated sender runtime"); + rt.block_on(swqos_worker_loop(queue, notify)); + }); + handles.push(handle); + } + let _ = DEDICATED_QUEUE.set(queue.clone()); + let _ = DEDICATED_NOTIFY.set(notify.clone()); + *guard = Some(handles); + (queue, notify) +} + fn ensure_swqos_pool(queue: Arc>) { if SWQOS_WORKERS_STARTED.swap(true, Ordering::AcqRel) { return; @@ -346,7 +403,7 @@ impl ResultCollector { pub async fn execute_parallel( swqos_clients: &[Arc], payer: Arc, - rpc: Option>, + rpc: Option<&Arc>, instructions: Vec, address_lookup_table_account: Option, recent_blockhash: Option, @@ -358,6 +415,8 @@ pub async fn execute_parallel( with_tip: bool, gas_fee_strategy: GasFeeStrategy, use_core_affinity: bool, + use_dedicated_sender_threads: bool, + sender_thread_cores: Option<&[usize]>, check_min_tip: bool, ) -> Result<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { let _exec_start = Instant::now(); @@ -427,7 +486,7 @@ pub async fn execute_parallel( let shared = Arc::new(SwqosSharedContext { payer, instructions, - rpc, + rpc: rpc.cloned(), address_lookup_table_account, recent_blockhash, durable_nonce, @@ -439,8 +498,13 @@ pub async fn execute_parallel( collector: collector.clone(), }); - let queue = SWQOS_QUEUE.get_or_init(|| Arc::new(ArrayQueue::new(SWQOS_QUEUE_CAP))); - ensure_swqos_pool(queue.clone()); + let (queue, notify) = if use_dedicated_sender_threads { + ensure_dedicated_pool(sender_thread_cores) + } else { + let q = SWQOS_QUEUE.get_or_init(|| Arc::new(ArrayQueue::new(SWQOS_QUEUE_CAP))); + ensure_swqos_pool(q.clone()); + (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. @@ -479,10 +543,7 @@ pub async fn execute_parallel( } } - // Wake all workers to process enqueued jobs - if let Some(notify) = SWQOS_NOTIFY.get() { - notify.notify_waiters(); - } + notify.notify_waiters(); // All jobs enqueued (no spawn on hot path) diff --git a/src/trading/core/executor.rs b/src/trading/core/executor.rs index 224d391..cc694de 100755 --- a/src/trading/core/executor.rs +++ b/src/trading/core/executor.rs @@ -145,9 +145,9 @@ impl TradeExecutor for GenericTradeExecutor { let need_confirm = params.wait_transaction_confirmed; let result = execute_parallel( - ¶ms.swqos_clients, + params.swqos_clients.as_slice(), params.payer, - params.rpc.clone(), + params.rpc.as_ref(), final_instructions, params.address_lookup_table_account, params.recent_blockhash, @@ -159,6 +159,8 @@ impl TradeExecutor for GenericTradeExecutor { 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()), params.check_min_tip, ) .await; @@ -171,6 +173,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 result = if need_confirm { diff --git a/src/trading/core/params.rs b/src/trading/core/params.rs index 93b57ed..5e27daa 100755 --- a/src/trading/core/params.rs +++ b/src/trading/core/params.rs @@ -56,7 +56,8 @@ pub struct SwapParams { pub wait_transaction_confirmed: bool, pub protocol_params: DexParamEnum, pub open_seed_optimize: bool, - pub swqos_clients: Vec>, + /// Arc> so cloning from infrastructure is a single Arc clone. + pub swqos_clients: Arc>>, pub middleware_manager: Option>, pub durable_nonce: Option, pub with_tip: bool, @@ -71,6 +72,10 @@ pub struct SwapParams { 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). + 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>>, /// 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. diff --git a/src/trading/core/transaction_pool.rs b/src/trading/core/transaction_pool.rs index f6a0e21..e3b57d3 100644 --- a/src/trading/core/transaction_pool.rs +++ b/src/trading/core/transaction_pool.rs @@ -12,8 +12,10 @@ 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; +/// 多路提交并发数（与 async_executor SWQOS_DEDICATED_DEFAULT_THREADS 一致，保证不串行） +const PARALLEL_SENDER_COUNT: usize = 18; +/// 启动时预填充数量，必须 >= PARALLEL_SENDER_COUNT，否则 18 路并发 build 会触发分配或争抢 +const TX_BUILDER_POOL_PREFILL: usize = 64; use crossbeam_queue::ArrayQueue; use once_cell::sync::Lazy; @@ -104,11 +106,10 @@ impl PreallocatedTxBuilder { /// 🚀 全局交易构建器对象池 static TX_BUILDER_POOL: Lazy>> = Lazy::new(|| { let pool = ArrayQueue::new(TX_BUILDER_POOL_CAP); - - for _ in 0..TX_BUILDER_POOL_PREFILL { + let prefill = TX_BUILDER_POOL_PREFILL.max(PARALLEL_SENDER_COUNT); + for _ in 0..prefill { let _ = pool.push(PreallocatedTxBuilder::new()); } - Arc::new(pool) });