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

perf(swqos): reduce submit latency and fix high latency after 5 min idle

Browse Source 
- Make serialize_transaction_and_encode sync and route through buffer pool
  (serialization::serialize_transaction_sync) to cut allocs on hot path
- Build single-submit body with format! instead of json!+to_string() in bloxroute
- Add serialize_transactions_batch_sync; bloxroute batch uses it and format! for entries
- Fix first-submit cold start: immediate first ping, then 30s keepalive interval
- Fix high latency after ~5 min: pool_max_idle_per_host=4, pool_idle_timeout=300s
  (BlockRazor, Temporal, Node1, Astralane, Stellium) so submit reuses ping connection
- Ping: 1.5s timeout, consume response body so connection returns to pool
- poll_transaction_confirmation: avoid cloning status.value[0], use ref
- Translate all swqos comments to English

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Wood
2026-02-20 02:18:02 +08:00
parent c27e479659
commit 5e4977f6a6
13 changed files with 182 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/swqos/astralane.rs
+14 -17
View File
@@ -52,8 +52,8 @@ impl AstralaneClient {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
// Optimized connection pool settings for high performance
.pool_idle_timeout(Duration::from_secs(120))
.pool_max_idle_per_host(256) // Increased from 64 to 256
.pool_idle_timeout(Duration::from_secs(300))
.pool_max_idle_per_host(4)
.tcp_keepalive(Some(Duration::from_secs(60))) // Reduced from 1200 to 60
.tcp_nodelay(true) // Disable Nagle's algorithm for lower latency
.http2_keep_alive_interval(Duration::from_secs(10))
@@ -90,17 +90,16 @@ impl AstralaneClient {
let stop_ping = self.stop_ping.clone();
let handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(60)); // Ping every 60 seconds
// Immediate first ping to warm connection and reduce first-submit cold start latency
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("Astralane ping request failed: {}", e);
}
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
// Send ping request
tokio::time::sleep(Duration::from_secs(5)).await;
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("Astralane ping request failed: {}", e);
}
@@ -130,25 +129,23 @@ impl AstralaneClient {
format!("{}/gethealth", endpoint)
};
// Send GET request to /gethealth endpoint with api_key header
// Short timeout for ping; consume body so connection is returned to pool for reuse by submit
let response = http_client.get(&ping_url)
.header("api_key", auth_token)
.timeout(Duration::from_millis(1500))
.send()
.await?;
if response.status().is_success() {
// ping successful, connection remains active
// println!("send getHealth to keep connection alive");
} else {
eprintln!("Astralane ping request returned non-success status: {}", response.status());
let status = response.status();
let _ = response.bytes().await;
if !status.is_success() {
eprintln!("Astralane ping request returned non-success status: {}", status);
}
Ok(())
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",
src/swqos/blockrazor.rs
+16 -18
View File
@@ -52,8 +52,8 @@ impl BlockRazorClient {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
// Optimized connection pool settings for high performance
.pool_idle_timeout(Duration::from_secs(120))
.pool_max_idle_per_host(256) // Increased from 64 to 256
.pool_idle_timeout(Duration::from_secs(300)) // 5min so ping-kept connection is not evicted early
.pool_max_idle_per_host(4) // Few connections so submit reuses same connection as ping, avoiding cold connection after ~5min server idle close
.tcp_keepalive(Some(Duration::from_secs(60))) // Reduced from 1200 to 60
.tcp_nodelay(true) // Disable Nagle's algorithm for lower latency
.http2_keep_alive_interval(Duration::from_secs(10))
@@ -90,16 +90,16 @@ impl BlockRazorClient {
let stop_ping = self.stop_ping.clone();
let handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(60)); // Ping every 60 seconds
// Immediate first ping to warm connection and reduce first-submit cold start latency
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("BlockRazor ping request failed: {}", e);
}
let mut interval = tokio::time::interval(Duration::from_secs(30)); // 30s keepalive to avoid server ~5min idle close
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
// Send ping request
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("BlockRazor ping request failed: {}", e);
}
@@ -137,33 +137,31 @@ impl BlockRazorClient {
headers.insert("apikey", HeaderValue::from_str(auth_token)?);
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
// Send GET request to /health endpoint with headers
// Short timeout for ping; consume body so connection is returned to pool for reuse by submit
let response = http_client.get(&ping_url)
.headers(headers)
.timeout(Duration::from_millis(1500))
.send()
.await?;
if response.status().is_success() {
// ping successful, connection remains active
// Can optionally log, but to reduce noise, not printing here
} else {
eprintln!("BlockRazor ping request failed with status: {}", response.status());
let status = response.status();
let _ = response.bytes().await;
if !status.is_success() {
eprintln!("BlockRazor ping request failed with status: {}", status);
}
Ok(())
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
// BlockRazor使用fast模式的请求格式
// BlockRazor fast-mode request format
let request_body = serde_json::to_string(&json!({
"transaction": content,
"mode": "fast"
}))?;
// BlockRazor使用apikey header
// BlockRazor uses apikey header
let response_text = self.http_client.post(&self.endpoint)
.body(request_body)
.header("Content-Type", "application/json")
src/swqos/bloxroute.rs
+24 -26
View File
@@ -1,4 +1,6 @@
use crate::swqos::common::{poll_transaction_confirmation, serialize_transaction_and_encode, FormatBase64VersionedTransaction};
use crate::swqos::common::poll_transaction_confirmation;
use crate::swqos::common::serialize_transaction_and_encode;
use crate::swqos::serialization;
use rand::seq::IndexedRandom;
use reqwest::Client;
use std::{sync::Arc, time::Instant};
@@ -63,27 +65,25 @@ impl BloxrouteClient {
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let body = serde_json::json!({
"transaction": {
"content": content,
},
"frontRunningProtection": false,
"useStakedRPCs": true,
});
// Single format! for body to avoid json! + to_string() double allocation
let body = format!(
r#"{{"transaction":{{"content":"{}"}},"frontRunningProtection":false,"useStakedRPCs":true}}"#,
content
);
let endpoint = format!("{}/api/v2/submit", self.endpoint);
let response_text = self.http_client.post(&endpoint)
.body(body.to_string())
.body(body)
.header("Content-Type", "application/json")
.header("Authorization", self.auth_token.clone())
.header("Authorization", self.auth_token.as_str())
.send()
.await?
.text()
.await?;
// 5. Use `serde_json::from_str()` to parse JSON, reducing extra wait from `.json().await?`
// Parse with from_str to avoid extra wait from .json().await
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" [bloxroute] {} submitted: {:?}", trade_type, start_time.elapsed());
@@ -114,24 +114,22 @@ impl BloxrouteClient {
pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>, _wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let body = serde_json::json!({
"entries": transactions
.iter()
.map(|tx| {
serde_json::json!({
"transaction": {
"content": tx.to_base64_string(),
},
})
})
.collect::<Vec<_>>(),
});
let contents = serialization::serialize_transactions_batch_sync(
transactions.as_slice(),
UiTransactionEncoding::Base64,
)?;
let entries: String = contents
.iter()
.map(|c| format!(r#"{{"transaction":{{"content":"{}"}}}}"#, c))
.collect::<Vec<_>>()
.join(",");
let body = format!(r#"{{"entries":[{}]}}"#, entries);
let endpoint = format!("{}/api/v2/submit-batch", self.endpoint);
let response_text = self.http_client.post(&endpoint)
.body(body.to_string())
.body(body)
.header("Content-Type", "application/json")
.header("Authorization", self.auth_token.clone())
.header("Authorization", self.auth_token.as_str())
.send()
.await?
.text()
src/swqos/common.rs
+21 -36
View File
@@ -3,6 +3,7 @@ use anyhow::Result;
use base64::engine::general_purpose::{self, STANDARD};
use base64::Engine;
use bincode::serialize;
use crate::swqos::serialization;
use reqwest::Client;
use serde_json;
use serde_json::json;
@@ -40,7 +41,7 @@ impl From<anyhow::Error> for TradeError {
}
}
// 使用高性能序列化
// High-performance serialization
pub trait FormatBase64VersionedTransaction {
fn to_base64_string(&self) -> String;
@@ -58,12 +59,12 @@ pub async fn poll_transaction_confirmation(
txt_sig: Signature,
wait_confirmation: bool,
) -> Result<Signature> {
// 如果不需要等待确认，立即返回签名
// If no confirmation needed, return signature immediately
if !wait_confirmation {
return Ok(txt_sig);
}
let timeout: Duration = Duration::from_secs(15); // 🔧 增加到15秒，避免网络拥堵时超时
let timeout: Duration = Duration::from_secs(15); // 15s to avoid timeout under network congestion
let interval: Duration = Duration::from_millis(1000);
let start: Instant = Instant::now();
let mut poll_count = 0u32;
@@ -76,33 +77,23 @@ pub async fn poll_transaction_confirmation(
poll_count += 1;
let status = rpc.get_signature_statuses(&[txt_sig]).await?;
match status.value[0].clone() {
Some(status) => {
if status.err.is_none()
&& (status.confirmation_status
== Some(TransactionConfirmationStatus::Confirmed)
|| status.confirmation_status
== Some(TransactionConfirmationStatus::Finalized))
let first = status.value.get(0).and_then(|o| o.as_ref());
match first {
Some(s) => {
if s.err.is_none()
&& (s.confirmation_status == Some(TransactionConfirmationStatus::Confirmed)
|| s.confirmation_status == Some(TransactionConfirmationStatus::Finalized))
{
return Ok(txt_sig);
}
// 如果 getSignatureStatuses 返回了错误，立即获取详细信息
if status.err.is_some() {
// 直接跳转到获取交易详情
}
}
None => {
// 交易还未上链，继续等待，不调用 getTransaction
sleep(interval).await;
continue;
}
}
// 优化：只在以下情况调用 getTransaction
// 1. getSignatureStatuses 返回了错误
// 2. 或者已经轮询了较长时间（超过10次，即10秒）
let should_get_transaction = status.value[0].as_ref().map(|s| s.err.is_some()).unwrap_or(false)
|| poll_count >= 10;
let should_get_transaction = first.map(|s| s.err.is_some()).unwrap_or(false) || poll_count >= 10;
if !should_get_transaction {
sleep(interval).await;
@@ -122,7 +113,7 @@ pub async fn poll_transaction_confirmation(
{
Ok(details) => details,
Err(_) => {
// 交易可能还未上链，继续等待
// Tx may not be on chain yet, keep waiting
sleep(interval).await;
continue;
}
@@ -136,7 +127,7 @@ pub async fn poll_transaction_confirmation(
if meta.err.is_none() {
return Ok(txt_sig);
} else {
// 从 log_messages 中提取错误信息
// Extract error message from log_messages
let mut error_msg = String::new();
if let solana_transaction_status::option_serializer::OptionSerializer::Some(logs) =
&meta.log_messages
@@ -162,12 +153,12 @@ pub async fn poll_transaction_confirmation(
let tx_err: TransactionError =
serde_json::from_value(serde_json::to_value(&ui_err)?)?;
// 直接使用Solana原生的InstructionError中的错误码
// Use Solana InstructionError codes directly
let mut code = 0u32;
let mut index = None;
match &tx_err {
TransactionError::InstructionError(i, i_error) => {
// 直接匹配所有InstructionError类型，Custom也是其中之一
// Match all InstructionError variants including Custom
code = match i_error {
solana_sdk::instruction::InstructionError::Custom(c) => *c,
solana_sdk::instruction::InstructionError::GenericError => 1,
@@ -180,7 +171,7 @@ pub async fn poll_transaction_confirmation(
solana_sdk::instruction::InstructionError::MissingRequiredSignature => 8,
solana_sdk::instruction::InstructionError::AccountAlreadyInitialized => 9,
solana_sdk::instruction::InstructionError::UninitializedAccount => 10,
_ => 999, // 其他未知错误
_ => 999, // Other unknown errors
};
index = Some(*i);
}
@@ -198,11 +189,11 @@ pub async fn poll_transaction_confirmation(
}
pub async fn send_nb_transaction(client: Client, endpoint: &str, auth_token: &str, transaction: &Transaction) -> Result<Signature, anyhow::Error> {
// 序列化交易
// Serialize transaction
let serialized = bincode::serialize(transaction)
.map_err(|e| anyhow::anyhow!("Transaction serialization failed: {}", e))?;
// Base64编码
// Base64 encode
let encoded = STANDARD.encode(serialized);
let request_data = json!({
@@ -250,18 +241,12 @@ pub async fn serialize_and_encode(
Ok(serialized)
}
pub async fn serialize_transaction_and_encode(
/// Sync serialize and encode; uses buffer pool when possible for lower allocs and latency.
pub fn serialize_transaction_and_encode(
transaction: &impl SerializableTransaction,
encoding: UiTransactionEncoding,
) -> Result<(String, Signature)> {
let signature = transaction.get_signature();
let serialized_tx = serialize(transaction)?;
let serialized = match encoding {
UiTransactionEncoding::Base58 => bs58::encode(serialized_tx).into_string(),
UiTransactionEncoding::Base64 => STANDARD.encode(serialized_tx),
_ => return Err(anyhow::anyhow!("Unsupported encoding")),
};
Ok((serialized, *signature))
serialization::serialize_transaction_sync(transaction, encoding)
}
pub async fn serialize_smart_transaction_and_encode(
src/swqos/flashblock.rs
+1 -1
View File
@@ -64,7 +64,7 @@ impl FlashBlockClient {
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
// FlashBlock API format
let request_body = serde_json::to_string(&json!({
src/swqos/jito.rs
+1 -1
View File
@@ -67,7 +67,7 @@ impl JitoClient {
pub async fn send_transaction_impl(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let request_body = serde_json::to_string(&json!({
"id": 1,
src/swqos/lightspeed.rs
+1 -1
View File
@@ -65,7 +65,7 @@ impl LightspeedClient {
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
// Lightspeed uses standard Solana JSON-RPC format for sendTransaction
let request_body = serde_json::to_string(&json!({
src/swqos/nextblock.rs
+1 -1
View File
@@ -69,7 +69,7 @@ impl NextBlockClient {
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let request_body = serde_json::to_string(&json!({
"transaction": {
src/swqos/node1.rs
+14 -16
View File
@@ -52,8 +52,8 @@ impl Node1Client {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
// Optimized connection pool settings for high performance
.pool_idle_timeout(Duration::from_secs(120))
.pool_max_idle_per_host(256) // Increased from 64 to 256
.pool_idle_timeout(Duration::from_secs(300))
.pool_max_idle_per_host(4)
.tcp_keepalive(Some(Duration::from_secs(60))) // Reduced from 1200 to 60
.tcp_nodelay(true) // Disable Nagle's algorithm for lower latency
.http2_keep_alive_interval(Duration::from_secs(10))
@@ -90,16 +90,16 @@ impl Node1Client {
let stop_ping = self.stop_ping.clone();
let handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(60)); // Ping every 60 seconds
// Immediate first ping to warm connection and reduce first-submit cold start latency
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("Node1 ping request failed: {}", e);
}
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
// Send ping request
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("Node1 ping request failed: {}", e);
}
@@ -125,24 +125,22 @@ impl Node1Client {
format!("{}/ping", endpoint)
};
// Send GET request to /ping endpoint (no api-key required)
// Short timeout for ping; consume body so connection is returned to pool for reuse by submit
let response = http_client.get(&ping_url)
.timeout(Duration::from_millis(1500))
.send()
.await?;
if response.status().is_success() {
// ping successful, connection remains active
// Can optionally log, but to reduce noise, not printing here
} else {
eprintln!("Node1 ping request returned non-success status: {}", response.status());
let status = response.status();
let _ = response.bytes().await;
if !status.is_success() {
eprintln!("Node1 ping request returned non-success status: {}", status);
}
Ok(())
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",
src/swqos/serialization.rs
+56 -21
View File
@@ -1,4 +1,4 @@
//! 交易序列化模块
//! Transaction serialization module.
use anyhow::Result;
use base64::Engine;
@@ -14,7 +14,7 @@ use crate::perf::{
compiler_optimization::CompileTimeOptimizedEventProcessor,
};
/// 零分配序列化器 - 使用缓冲池避免运行时分配
/// Zero-allocation serializer using a buffer pool to avoid runtime allocation.
pub struct ZeroAllocSerializer {
buffer_pool: Arc<ArrayQueue<Vec<u8>>>,
buffer_size: usize,
@@ -24,7 +24,7 @@ impl ZeroAllocSerializer {
pub fn new(pool_size: usize, buffer_size: usize) -> Self {
let pool = ArrayQueue::new(pool_size);
// 预分配缓冲区
// Pre-allocate buffers
for _ in 0..pool_size {
let mut buffer = Vec::with_capacity(buffer_size);
buffer.resize(buffer_size, 0);
@@ -38,14 +38,14 @@ impl ZeroAllocSerializer {
}
pub fn serialize_zero_alloc<T: serde::Serialize>(&self, data: &T, _label: &str) -> Result<Vec<u8>> {
// 尝试从池中获取缓冲区
// Try to get a buffer from the pool
let mut buffer = self.buffer_pool.pop().unwrap_or_else(|| {
let mut buf = Vec::with_capacity(self.buffer_size);
buf.resize(self.buffer_size, 0);
buf
});
// 序列化到缓冲区
// Serialize into buffer
let serialized = bincode::serialize(data)?;
buffer.clear();
buffer.extend_from_slice(&serialized);
@@ -54,11 +54,11 @@ impl ZeroAllocSerializer {
}
pub fn return_buffer(&self, buffer: Vec<u8>) {
// 归还缓冲区到池中
// Return buffer to the pool
let _ = self.buffer_pool.push(buffer);
}
/// 获取池统计信息
/// Get pool statistics.
pub fn get_pool_stats(&self) -> (usize, usize) {
let available = self.buffer_pool.len();
let capacity = self.buffer_pool.capacity();
@@ -66,32 +66,32 @@ impl ZeroAllocSerializer {
}
}
/// 全局序列化器实例
/// Global serializer instance.
static SERIALIZER: Lazy<Arc<ZeroAllocSerializer>> = Lazy::new(|| {
Arc::new(ZeroAllocSerializer::new(
10_000, // 池大小
256 * 1024, // 缓冲区大小: 256KB
10_000, // Pool size
256 * 1024, // Buffer size: 256KB
))
});
/// 🚀 编译时优化的事件处理器 (零运行时开销)
/// Compile-time optimized event processor (zero runtime cost).
static COMPILE_TIME_PROCESSOR: CompileTimeOptimizedEventProcessor =
CompileTimeOptimizedEventProcessor::new();
/// Base64 编码器
/// Base64 encoder.
pub struct Base64Encoder;
impl Base64Encoder {
#[inline(always)]
pub fn encode(data: &[u8]) -> String {
// 使用编译时优化的哈希进行快速路由
// Use compile-time optimized hash for fast routing
let _route = if !data.is_empty() {
COMPILE_TIME_PROCESSOR.route_event_zero_cost(data[0])
} else {
0
};
// 使用 SIMD 加速的 Base64 编码
// Use SIMD-accelerated Base64 encoding
SIMDSerializer::encode_base64_simd(data)
}
@@ -105,32 +105,67 @@ impl Base64Encoder {
}
}
/// 交易序列化
/// Sync serialize + encode using buffer pool; use in hot path to reduce allocs.
pub fn serialize_transaction_sync(
transaction: &impl SerializableTransaction,
encoding: UiTransactionEncoding,
) -> Result<(String, Signature)> {
let signature = transaction.get_signature();
let serialized_tx = SERIALIZER.serialize_zero_alloc(transaction, "transaction")?;
let serialized = match encoding {
UiTransactionEncoding::Base58 => bs58::encode(&serialized_tx).into_string(),
UiTransactionEncoding::Base64 => STANDARD.encode(&serialized_tx),
_ => return Err(anyhow::anyhow!("Unsupported encoding")),
};
SERIALIZER.return_buffer(serialized_tx);
Ok((serialized, *signature))
}
/// Serialize a transaction (async; no I/O, kept for API compatibility).
pub async fn serialize_transaction(
transaction: &impl SerializableTransaction,
encoding: UiTransactionEncoding,
) -> Result<(String, Signature)> {
let signature = transaction.get_signature();
// 使用零分配序列化
// Use zero-allocation serialization
let serialized_tx = SERIALIZER.serialize_zero_alloc(transaction, "transaction")?;
let serialized = match encoding {
UiTransactionEncoding::Base58 => bs58::encode(&serialized_tx).into_string(),
UiTransactionEncoding::Base64 => {
// 使用 SIMD 优化的 Base64 编码
// Use SIMD-optimized Base64 encoding
STANDARD.encode(&serialized_tx)
}
_ => return Err(anyhow::anyhow!("Unsupported encoding")),
};
// 立即归还缓冲区到池中
// Return buffer to pool immediately
SERIALIZER.return_buffer(serialized_tx);
Ok((serialized, *signature))
}
/// 批量交易序列化
/// Sync batch serialize + encode using buffer pool.
pub fn serialize_transactions_batch_sync(
transactions: &[impl SerializableTransaction],
encoding: UiTransactionEncoding,
) -> Result<Vec<String>> {
let mut results = Vec::with_capacity(transactions.len());
for tx in transactions {
let serialized_tx = SERIALIZER.serialize_zero_alloc(tx, "transaction")?;
let encoded = match encoding {
UiTransactionEncoding::Base58 => bs58::encode(&serialized_tx).into_string(),
UiTransactionEncoding::Base64 => STANDARD.encode(&serialized_tx),
_ => return Err(anyhow::anyhow!("Unsupported encoding")),
};
SERIALIZER.return_buffer(serialized_tx);
results.push(encoded);
}
Ok(results)
}
/// Batch transaction serialization.
pub async fn serialize_transactions_batch(
transactions: &[impl SerializableTransaction],
encoding: UiTransactionEncoding,
@@ -153,7 +188,7 @@ pub async fn serialize_transactions_batch(
Ok(results)
}
/// 获取序列化器统计信息
/// Get serializer statistics.
pub fn get_serializer_stats() -> (usize, usize) {
SERIALIZER.get_pool_stats()
}
@@ -168,7 +203,7 @@ mod tests {
let encoded = Base64Encoder::encode(data);
assert!(!encoded.is_empty());
// 验证可以正确解码
// Verify it decodes correctly
let decoded = STANDARD.decode(&encoded).unwrap();
assert_eq!(&decoded[..data.len()], data);
}
src/swqos/stellium.rs
+18 -11
View File
@@ -50,8 +50,8 @@ impl StelliumClient {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
// Optimized connection pool settings for high performance
.pool_idle_timeout(Duration::from_secs(120))
.pool_max_idle_per_host(256) // Increased from 64 to 256
.pool_idle_timeout(Duration::from_secs(300))
.pool_max_idle_per_host(4)
.tcp_keepalive(Some(Duration::from_secs(60))) // Reduced from 1200 to 60
.tcp_nodelay(true) // Disable Nagle's algorithm for lower latency
.http2_keep_alive_interval(Duration::from_secs(10))
@@ -89,21 +89,28 @@ impl StelliumClient {
let stop_ping = self.keep_alive_running.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(60)); // Ping every 60 seconds
// Immediate first ping to warm connection and reduce first-submit cold start latency
let url = format!("{}/{}", endpoint, auth_token);
if let Ok(resp) = http_client.get(&url).timeout(Duration::from_millis(1500)).send().await {
let status = resp.status();
let _ = resp.bytes().await;
if !status.is_success() {
eprintln!(" [Stellium] Ping failed with status: {}", status);
}
}
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
// Send ping request
let url = format!("{}/{}", endpoint, auth_token);
match http_client.get(&url).send().await {
match http_client.get(&url).timeout(Duration::from_millis(1500)).send().await {
Ok(response) => {
if !response.status().is_success() {
eprintln!(" [Stellium] Ping failed with status: {}", response.status());
let status = response.status();
let _ = response.bytes().await;
if !status.is_success() {
eprintln!(" [Stellium] Ping failed with status: {}", status);
}
}
Err(e) => {
@@ -116,7 +123,7 @@ impl StelliumClient {
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
// Stellium uses standard Solana sendTransaction format
let request_body = serde_json::to_string(&json!({
src/swqos/temporal.rs
+14 -16
View File
@@ -78,8 +78,8 @@ impl TemporalClient {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
// Optimized connection pool settings for high performance
.pool_idle_timeout(Duration::from_secs(120))
.pool_max_idle_per_host(256) // Increased from 64 to 256
.pool_idle_timeout(Duration::from_secs(300))
.pool_max_idle_per_host(4)
.tcp_keepalive(Some(Duration::from_secs(60))) // Reduced from 1200 to 60
.tcp_nodelay(true) // Disable Nagle's algorithm for lower latency
.http2_keep_alive_interval(Duration::from_secs(10))
@@ -116,16 +116,16 @@ impl TemporalClient {
let stop_ping = self.stop_ping.clone();
let handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(60)); // Ping every 60 seconds
// Immediate first ping to warm connection and reduce first-submit cold start latency
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("Temporal ping request failed: {}", e);
}
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
interval.tick().await;
if stop_ping.load(Ordering::Relaxed) {
break;
}
// Send ping request
if let Err(e) = Self::send_ping_request(&http_client, &endpoint, &auth_token).await {
eprintln!("Temporal ping request failed: {}", e);
}
@@ -151,24 +151,22 @@ impl TemporalClient {
format!("{}/ping", endpoint)
};
// Send GET request to /ping endpoint
// Short timeout for ping; consume body so connection is returned to pool for reuse by submit
let response = http_client.get(&ping_url)
.timeout(Duration::from_millis(1500))
.send()
.await?;
if response.status().is_success() {
// ping successful, connection remains active
// Can optionally log, but to reduce noise, not printing here
} else {
eprintln!("Temporal ping request returned non-success status: {}", response.status());
let status = response.status();
let _ = response.bytes().await;
if !status.is_success() {
eprintln!("Temporal ping request returned non-success status: {}", status);
}
Ok(())
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
// Build request body according to Nozomi documentation requirements
let request_body = serde_json::to_string(&json!({
src/swqos/zeroslot.rs
+1 -1
View File
@@ -64,7 +64,7 @@ impl ZeroSlotClient {
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction, wait_confirmation: bool) -> Result<()> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64)?;
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",