diff --git a/src/trading/core/async_executor.rs b/src/trading/core/async_executor.rs index 24f53eb..149eaa7 100644 --- a/src/trading/core/async_executor.rs +++ b/src/trading/core/async_executor.rs @@ -22,9 +22,10 @@ struct TaskResult { success: bool, signature: Signature, error: Option, - #[allow(dead_code)] swqos_type: SwqosType, landed_on_chain: bool, + /// Microsecond timestamp when this task finished (SWQOS returned); for per-SWQOS event→submit timing. + submit_done_us: i64, } /// Check if an error indicates the transaction landed on-chain (vs network/timeout error) @@ -87,7 +88,7 @@ impl ResultCollector { self.completed_count.fetch_add(1, Ordering::Release); } - async fn wait_for_success(&self) -> Option<(bool, Vec, Option)> { + async fn wait_for_success(&self) -> Option<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { let start = Instant::now(); let timeout = std::time::Duration::from_secs(5); let poll_interval = std::time::Duration::from_millis(1000); @@ -96,14 +97,16 @@ impl ResultCollector { if self.success_flag.load(Ordering::Acquire) { let mut signatures = Vec::new(); let mut has_success = false; + let mut submit_timings = Vec::new(); while let Some(result) = self.results.pop() { signatures.push(result.signature); + submit_timings.push((result.swqos_type, result.submit_done_us)); if result.success { has_success = true; } } if has_success && !signatures.is_empty() { - return Some((true, signatures, None)); + return Some((true, signatures, None, submit_timings)); } } @@ -112,15 +115,17 @@ impl ResultCollector { if self.landed_failed_flag.load(Ordering::Acquire) { let mut signatures = Vec::new(); let mut landed_error = None; + let mut submit_timings = Vec::new(); while let Some(result) = self.results.pop() { signatures.push(result.signature); + submit_timings.push((result.swqos_type, result.submit_done_us)); // Prefer the error from the tx that actually landed if result.landed_on_chain && result.error.is_some() { landed_error = result.error; } } if !signatures.is_empty() { - return Some((false, signatures, landed_error)); + return Some((false, signatures, landed_error, submit_timings)); } } @@ -129,8 +134,10 @@ impl ResultCollector { let mut signatures = Vec::new(); let mut last_error = None; let mut any_success = false; + let mut submit_timings = Vec::new(); while let Some(result) = self.results.pop() { signatures.push(result.signature); + submit_timings.push((result.swqos_type, result.submit_done_us)); if result.success { any_success = true; } @@ -139,7 +146,7 @@ impl ResultCollector { } } if !signatures.is_empty() { - return Some((any_success, signatures, last_error)); + return Some((any_success, signatures, last_error, submit_timings)); } return None; } @@ -151,13 +158,15 @@ impl ResultCollector { } } - fn get_first(&self) -> Option<(bool, Vec, Option)> { + fn get_first(&self) -> Option<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { let mut signatures = Vec::new(); let mut has_success = false; let mut last_error = None; - + let mut submit_timings = Vec::new(); + while let Some(result) = self.results.pop() { signatures.push(result.signature); + submit_timings.push((result.swqos_type, result.submit_done_us)); if result.success { has_success = true; } @@ -165,19 +174,20 @@ impl ResultCollector { last_error = result.error; } } - + if !signatures.is_empty() { - Some((has_success, signatures, last_error)) + Some((has_success, signatures, last_error, submit_timings)) } else { None } } /// 等待全部任务完成（不等待链上确认），然后收集并返回所有签名。用于「多路提交」时返回多笔签名。 - async fn wait_for_all_submitted(&self, timeout_secs: u64) -> Option<(bool, Vec, Option)> { + /// 轮询间隔 2ms，避免 50ms 间隔在最后一笔返回时多等几十 ms 拉高 submit 耗时。 + async fn wait_for_all_submitted(&self, timeout_secs: u64) -> Option<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { let start = Instant::now(); let timeout = std::time::Duration::from_secs(timeout_secs); - let poll_interval = std::time::Duration::from_millis(50); + let poll_interval = std::time::Duration::from_millis(2); while self.completed_count.load(Ordering::Acquire) < self.total_tasks { if start.elapsed() > timeout { break; @@ -205,7 +215,7 @@ pub async fn execute_parallel( gas_fee_strategy: GasFeeStrategy, use_core_affinity: bool, check_min_tip: bool, -) -> Result<(bool, Vec, Option)> { +) -> Result<(bool, Vec, Option, Vec<(SwqosType, i64)>)> { let _exec_start = Instant::now(); if swqos_clients.is_empty() { @@ -334,6 +344,7 @@ pub async fn execute_parallel( error: Some(e), swqos_type, landed_on_chain: false, + submit_done_us: crate::common::clock::now_micros(), }); return; } @@ -375,6 +386,7 @@ pub async fn execute_parallel( error: err, swqos_type, landed_on_chain, + submit_done_us: crate::common::clock::now_micros(), }); }); } @@ -383,15 +395,17 @@ pub async fn execute_parallel( if !wait_transaction_confirmed { const SUBMIT_TIMEOUT_SECS: u64 = 30; - let (success, signatures, last_error) = collector + let ret = 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)); + .unwrap_or((false, vec![], Some(anyhow!("No SWQOS result within {}s", SUBMIT_TIMEOUT_SECS)), vec![])); + let (success, signatures, last_error, submit_timings) = ret; + return Ok((success, signatures, last_error, submit_timings)); } if let Some(result) = collector.wait_for_success().await { - Ok(result) + let (success, signatures, last_error, submit_timings) = result; + Ok((success, signatures, last_error, submit_timings)) } else { Err(anyhow!("All transactions failed")) } diff --git a/src/trading/core/executor.rs b/src/trading/core/executor.rs index 9f565f5..89b1372 100755 --- a/src/trading/core/executor.rs +++ b/src/trading/core/executor.rs @@ -69,7 +69,7 @@ impl TradeExecutor for GenericTradeExecutor { } else { self.instruction_builder.build_sell_instructions(¶ms).await? }; - let build_elapsed = build_start.map(|s| s.elapsed()).unwrap_or(Duration::ZERO); + let _build_elapsed = build_start.map(|s| s.elapsed()).unwrap_or(Duration::ZERO); InstructionProcessor::preprocess(&instructions)?; @@ -83,7 +83,11 @@ impl TradeExecutor for GenericTradeExecutor { None => instructions, }; - let before_submit_elapsed = total_start.as_ref().map(|s| s.elapsed()).unwrap_or(Duration::ZERO); + let build_end_us = (params.log_enabled && crate::common::sdk_log::sdk_log_enabled()) + .then(crate::common::clock::now_micros); + let _before_submit_elapsed = total_start.as_ref().map(|s| s.elapsed()).unwrap_or(Duration::ZERO); + let before_submit_us = (params.log_enabled && crate::common::sdk_log::sdk_log_enabled()) + .then(crate::common::clock::now_micros); if params.simulate { let send_start = crate::common::sdk_log::sdk_log_enabled().then(Instant::now); @@ -106,14 +110,20 @@ impl TradeExecutor for GenericTradeExecutor { 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); + if let (Some(start_us), Some(end_us)) = (timing_start_us, build_end_us) { + println!(" [SDK] {} build_instructions: {:.4} ms", dir, (end_us - start_us) as f64 / 1000.0); + } + if let (Some(start_us), Some(end_us)) = (timing_start_us, before_submit_us) { + println!(" [SDK] {} before_submit: {:.4} ms", dir, (end_us - start_us) as f64 / 1000.0); + } + println!(" [SDK] {} simulate (dry-run): {:.4} ms", dir, send_elapsed.as_secs_f64() * 1000.0); + println!(" [SDK] {} total: {:.4} ms", dir, total_elapsed.as_secs_f64() * 1000.0); } return result; } let need_confirm = params.wait_transaction_confirmed; - let send_start = params.log_enabled.then(Instant::now); let result = execute_parallel( ¶ms.swqos_clients, params.payer, @@ -132,25 +142,17 @@ impl TradeExecutor for GenericTradeExecutor { params.check_min_tip, ) .await; - let send_elapsed = send_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 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); - } - } + 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 result = if need_confirm { let (ok, sigs, err) = match &result { - Ok((success, signatures, last_error)) => ( + Ok((success, signatures, last_error, _)) => ( *success, signatures.clone(), last_error.as_ref().map(|e| anyhow::anyhow!("{}", e)), @@ -161,14 +163,26 @@ impl TradeExecutor for GenericTradeExecutor { if sigs.is_empty() { (ok, sigs, err) } else { - let confirm_start = (params.log_enabled && crate::common::sdk_log::sdk_log_enabled()).then(Instant::now); let poll_res = poll_any_transaction_confirmation(rpc, &sigs, 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 confirm_done_us = log_enabled.then(crate::common::clock::now_micros); + if 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); + 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); + } + } + } } match poll_res { Ok(_) => (true, sigs, None), @@ -180,12 +194,22 @@ impl TradeExecutor for GenericTradeExecutor { }; 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; + if log_enabled { let dir = if is_buy { "Buy" } else { "Sell" }; - println!(" [SDK] {} timing total: {:.2}ms", dir, total_ms); + 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); + } + for (swqos_type, submit_done_us) in &submit_timings { + 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 + result.map(|(a, b, c, _)| (a, b, c)) }; result @@ -304,3 +328,49 @@ async fn simulate_transaction( Ok((true, vec![signature], None)) } + +#[cfg(test)] +mod tests { + use crate::swqos::SwqosType; + + /// 运行 `cargo test -p sol-trade-sdk log_timing_preview -- --nocapture` 查看日志打印效果 + #[test] + fn log_timing_preview() { + let dir = "Buy"; + let build_ms = 12.34; + let before_submit_ms = 15.67; + println!("\n--- 1. 构建指令耗时 / 提交前耗时（各打印一次，统一 ms，保留 4 位小数）---\n"); + println!(" [SDK] {} build_instructions: {:.4} ms", dir, build_ms); + println!(" [SDK] {} before_submit: {:.4} ms", dir, before_submit_ms); + + println!("\n--- 2. 每个 SWQOS 独立耗时：submit=起点→该通道返回, confirmed=该通道提交→链上确认, total=起点→链上确认 ---\n"); + for (swqos_type, submit_ms, confirmed_ms, total_ms) in [ + (SwqosType::Jito, 45.12, 83.38, 128.50), + (SwqosType::Helius, 52.30, 76.20, 128.50), + (SwqosType::ZeroSlot, 48.90, 79.60, 128.50), + ] { + println!( + " [SDK] {} {:?} submit: {:.4} ms, confirmed: {:.4} ms, total: {:.4} ms", + dir, swqos_type, submit_ms, confirmed_ms, total_ms + ); + } + + println!("\n--- 3. 不等待链上确认时：每行 total = 该通道 submit 耗时（独立）---\n"); + for (swqos_type, submit_ms, total_ms) in [ + (SwqosType::Jito, 44.20, 44.20), + (SwqosType::Helius, 51.80, 51.80), + ] { + println!( + " [SDK] {} {:?} submit: {:.4} ms, confirmed: -, total: {:.4} ms", + dir, swqos_type, submit_ms, total_ms + ); + } + + println!("\n--- 4. Simulate 模式（build/before_submit 仍从 grpc_recv_us 起算）---\n"); + println!(" [SDK] {} build_instructions: {:.4} ms", dir, build_ms); + println!(" [SDK] {} before_submit: {:.4} ms", dir, before_submit_ms); + println!(" [SDK] {} simulate (dry-run): {:.4} ms", dir, 8.50); + println!(" [SDK] {} total: {:.4} ms", dir, 36.51); + println!(); + } +}