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

@@ -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)

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，无每次加锁。

examples/middleware_system/src/main.rs

@@ -31,7 +31,7 @@ impl InstructionMiddleware for CustomMiddleware {
     fn process_protocol_instructions(
         &self,
         protocol_instructions: Vec<Instruction>,
-        _protocol_name: String,
+        _protocol_name: &str,
         _is_buy: bool,
     ) -> Result<Vec<Instruction>> {
         // do anything you want here
@@ -42,7 +42,7 @@ impl InstructionMiddleware for CustomMiddleware {
     fn process_full_instructions(
         &self,
         full_instructions: Vec<Instruction>,
-        _protocol_name: String,
+        _protocol_name: &str,
         _is_buy: bool,
     ) -> Result<Vec<Instruction>> {
         // do anything you want here

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<std::sync::Arc<Vec<usize>>>,
     /// 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
         }

src/lib.rs

@@ -86,8 +86,8 @@ pub enum TradeTokenType {
 pub struct TradingInfrastructure {
     /// Shared RPC client for blockchain interactions
     pub rpc: Arc<SolanaRpcClient>,
-    /// Shared SWQOS clients for transaction priority and routing
-    pub swqos_clients: Vec<Arc<SwqosClient>>,
+    /// Shared SWQOS clients for transaction priority and routing. Arc<Vec<..>> so cloning into SwapParams is a single Arc clone.
+    pub swqos_clients: Arc<Vec<Arc<SwqosClient>>>,
     /// 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<Arc<Vec<usize>>>,
     /// 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,

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<Arc<ArrayQueue<SwqosJob>>> = OnceCell::new();
 static SWQOS_NOTIFY: OnceCell<Arc<Notify>> = 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<Arc<ArrayQueue<SwqosJob>>> = OnceCell::new();
+static DEDICATED_NOTIFY: OnceCell<Arc<Notify>> = OnceCell::new();
+/// JoinHandles kept so dedicated threads are not detached; only touched during init under lock.
+static DEDICATED_INIT: Mutex<Option<Vec<std::thread::JoinHandle<()>>>> = Mutex::new(None);
+
+fn ensure_dedicated_pool(sender_thread_cores: Option<&[usize]>) -> (Arc<ArrayQueue<SwqosJob>>, Arc<Notify>) {
+    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<core_affinity::CoreId> = 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<ArrayQueue<SwqosJob>>) {
     if SWQOS_WORKERS_STARTED.swap(true, Ordering::AcqRel) {
         return;
@@ -346,7 +403,7 @@ impl ResultCollector {
 pub async fn execute_parallel(
     swqos_clients: &[Arc<SwqosClient>],
     payer: Arc<Keypair>,
-    rpc: Option<Arc<SolanaRpcClient>>,
+    rpc: Option<&Arc<SolanaRpcClient>>,
     instructions: Vec<Instruction>,
     address_lookup_table_account: Option<AddressLookupTableAccount>,
     recent_blockhash: Option<Hash>,
@@ -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<Signature>, Option<anyhow::Error>, 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)
 

src/trading/core/executor.rs

@@ -145,9 +145,9 @@ impl TradeExecutor for GenericTradeExecutor {
 
         let need_confirm = params.wait_transaction_confirmed;
         let result = execute_parallel(
-            &params.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 {

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<SwqosClient>>,
+    /// Arc<Vec<..>> so cloning from infrastructure is a single Arc clone.
+    pub swqos_clients: Arc<Vec<Arc<SwqosClient>>>,
     pub middleware_manager: Option<Arc<MiddlewareManager>>,
     pub durable_nonce: Option<DurableNonceInfo>,
     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<Arc<Vec<usize>>>,
     /// 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.

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<Arc<ArrayQueue<PreallocatedTxBuilder>>> = 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)
 });