- Bump version to 3.5.0 - Performance: hot-path timing only when log_enabled/simulate; execute_parallel takes &[Arc<SwqosClient>]; shared HTTP client constants for SWQoS - Code quality: validate_protocol_params extracted for buy/sell; BYTES_PER_ACCOUNT, MAX_INSTRUCTIONS_WARN, HTTP timeout constants; prefetch/syscall bypass comments - Documentation: bilingual (EN + 中文) doc comments in execution, executor, perf, swqos; README/README_CN version and What's new in 3.5.0 - Add release_notes_v3.5.0.md Co-authored-by: Cursor <cursoragent@cursor.com>
746 lines
25 KiB
Rust
746 lines
25 KiB
Rust
//! Syscall bypass: batching, vDSO fast time, io_uring, mmap, userspace impl.
|
|
//! 系统调用绕过:批处理、vDSO 快速时间、io_uring、mmap、用户态实现。
|
|
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::Arc;
|
|
use std::time::{SystemTime, UNIX_EPOCH, Duration, Instant};
|
|
#[allow(unused_imports)]
|
|
use std::fs::OpenOptions;
|
|
|
|
use anyhow::Result;
|
|
use crossbeam_utils::CachePadded;
|
|
|
|
/// Syscall bypass manager (batch, fast time, I/O). 系统调用绕过管理器。
|
|
pub struct SystemCallBypassManager {
|
|
config: SyscallBypassConfig,
|
|
batch_processor: Arc<SyscallBatchProcessor>,
|
|
fast_time_provider: Arc<FastTimeProvider>,
|
|
_io_optimizer: Arc<IOOptimizer>,
|
|
stats: Arc<SyscallBypassStats>,
|
|
}
|
|
|
|
/// Syscall bypass configuration. 系统调用绕过配置。
|
|
#[derive(Debug, Clone)]
|
|
pub struct SyscallBypassConfig {
|
|
pub enable_batch_processing: bool,
|
|
pub batch_size: usize,
|
|
pub enable_fast_time: bool,
|
|
pub enable_vdso: bool,
|
|
pub enable_io_uring: bool,
|
|
pub enable_mmap_optimization: bool,
|
|
pub enable_userspace_impl: bool,
|
|
pub syscall_cache_size: usize,
|
|
}
|
|
|
|
impl Default for SyscallBypassConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
enable_batch_processing: true,
|
|
batch_size: 100,
|
|
enable_fast_time: true,
|
|
enable_vdso: true,
|
|
enable_io_uring: true,
|
|
enable_mmap_optimization: true,
|
|
enable_userspace_impl: true,
|
|
syscall_cache_size: 1000,
|
|
}
|
|
}
|
|
}
|
|
|
|
pub struct SyscallBatchProcessor {
|
|
pending_calls: crossbeam_queue::ArrayQueue<SyscallRequest>,
|
|
_executor: tokio::runtime::Handle,
|
|
batch_stats: CachePadded<AtomicU64>,
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub enum SyscallRequest {
|
|
Write { fd: i32, data: Vec<u8> },
|
|
Read { fd: i32, size: usize },
|
|
Send { socket: i32, data: Vec<u8> },
|
|
Recv { socket: i32, size: usize },
|
|
GetTime,
|
|
MemAlloc { size: usize },
|
|
/// 内存释放
|
|
MemFree { ptr: usize },
|
|
}
|
|
|
|
/// 🚀 快速时间提供器 - 绕过系统调用获取时间
|
|
pub struct FastTimeProvider {
|
|
/// 时间基准点
|
|
_base_time: SystemTime,
|
|
/// 单调时间起始点
|
|
monotonic_start: Instant,
|
|
/// 时间缓存
|
|
time_cache: CachePadded<AtomicU64>,
|
|
/// 缓存更新间隔 (纳秒)
|
|
cache_update_interval_ns: u64,
|
|
/// 上次更新时间
|
|
last_update: CachePadded<AtomicU64>,
|
|
/// 启用vDSO
|
|
vdso_enabled: bool,
|
|
}
|
|
|
|
impl FastTimeProvider {
|
|
/// 创建快速时间提供器
|
|
pub fn new(enable_vdso: bool) -> Result<Self> {
|
|
let now = SystemTime::now();
|
|
let instant_now = Instant::now();
|
|
|
|
let provider = Self {
|
|
_base_time: now,
|
|
monotonic_start: instant_now,
|
|
time_cache: CachePadded::new(AtomicU64::new(
|
|
now.duration_since(UNIX_EPOCH)?.as_nanos() as u64
|
|
)),
|
|
cache_update_interval_ns: 1_000_000, // 1ms
|
|
last_update: CachePadded::new(AtomicU64::new(
|
|
instant_now.elapsed().as_nanos() as u64
|
|
)),
|
|
vdso_enabled: enable_vdso,
|
|
};
|
|
|
|
tracing::info!(target: "sol_trade_sdk","🚀 Fast time provider initialized with vDSO: {}", enable_vdso);
|
|
Ok(provider)
|
|
}
|
|
|
|
/// 🚀 超快速获取当前时间 - 绕过系统调用
|
|
#[inline(always)]
|
|
pub fn fast_now_nanos(&self) -> u64 {
|
|
if self.vdso_enabled {
|
|
// 使用vDSO快速获取时间
|
|
return self.vdso_time_nanos();
|
|
}
|
|
|
|
// 使用缓存的时间
|
|
let now_mono = self.monotonic_start.elapsed().as_nanos() as u64;
|
|
let last_update = self.last_update.load(Ordering::Relaxed);
|
|
|
|
if now_mono.saturating_sub(last_update) > self.cache_update_interval_ns {
|
|
// 需要更新缓存
|
|
self.update_time_cache();
|
|
}
|
|
|
|
self.time_cache.load(Ordering::Relaxed)
|
|
}
|
|
|
|
/// vDSO时间获取
|
|
#[inline(always)]
|
|
fn vdso_time_nanos(&self) -> u64 {
|
|
#[cfg(target_os = "linux")]
|
|
{
|
|
// 在Linux上使用vDSO获取时间,避免系统调用
|
|
unsafe {
|
|
let mut ts = libc::timespec { tv_sec: 0, tv_nsec: 0 };
|
|
|
|
// CLOCK_MONOTONIC_RAW不受NTP调整影响,更适合性能测量
|
|
if libc::clock_gettime(libc::CLOCK_MONOTONIC_RAW, &mut ts) == 0 {
|
|
return (ts.tv_sec as u64) * 1_000_000_000 + (ts.tv_nsec as u64);
|
|
}
|
|
}
|
|
}
|
|
|
|
// 回退到缓存时间
|
|
self.time_cache.load(Ordering::Relaxed)
|
|
}
|
|
|
|
/// 更新时间缓存
|
|
fn update_time_cache(&self) {
|
|
if let Ok(now) = SystemTime::now().duration_since(UNIX_EPOCH) {
|
|
let nanos = now.as_nanos() as u64;
|
|
self.time_cache.store(nanos, Ordering::Relaxed);
|
|
self.last_update.store(
|
|
self.monotonic_start.elapsed().as_nanos() as u64,
|
|
Ordering::Relaxed
|
|
);
|
|
}
|
|
}
|
|
|
|
/// 🚀 快速获取微秒时间戳
|
|
#[inline(always)]
|
|
pub fn fast_now_micros(&self) -> u64 {
|
|
self.fast_now_nanos() / 1000
|
|
}
|
|
|
|
/// 🚀 快速获取毫秒时间戳
|
|
#[inline(always)]
|
|
pub fn fast_now_millis(&self) -> u64 {
|
|
self.fast_now_nanos() / 1_000_000
|
|
}
|
|
}
|
|
|
|
/// 🚀 I/O优化器 - 使用io_uring等高性能I/O
|
|
pub struct IOOptimizer {
|
|
/// io_uring是否可用
|
|
io_uring_available: bool,
|
|
/// 异步I/O统计
|
|
async_io_stats: Arc<AsyncIOStats>,
|
|
/// 内存映射区域
|
|
mmap_regions: Vec<MemoryMappedRegion>,
|
|
}
|
|
|
|
/// 异步I/O统计
|
|
#[derive(Debug, Default)]
|
|
pub struct AsyncIOStats {
|
|
pub operations_queued: AtomicU64,
|
|
pub operations_completed: AtomicU64,
|
|
pub bytes_transferred: AtomicU64,
|
|
pub syscalls_avoided: AtomicU64,
|
|
}
|
|
|
|
/// 内存映射区域
|
|
#[derive(Debug)]
|
|
pub struct MemoryMappedRegion {
|
|
pub address: usize,
|
|
pub size: usize,
|
|
pub file_descriptor: i32,
|
|
}
|
|
|
|
impl IOOptimizer {
|
|
/// 创建I/O优化器
|
|
pub fn new(_config: &SyscallBypassConfig) -> Result<Self> {
|
|
let io_uring_available = Self::check_io_uring_support();
|
|
|
|
tracing::info!(target: "sol_trade_sdk","🚀 I/O Optimizer initialized - io_uring: {}", io_uring_available);
|
|
|
|
Ok(Self {
|
|
io_uring_available,
|
|
async_io_stats: Arc::new(AsyncIOStats::default()),
|
|
mmap_regions: Vec::new(),
|
|
})
|
|
}
|
|
|
|
/// 检查io_uring支持
|
|
fn check_io_uring_support() -> bool {
|
|
#[cfg(target_os = "linux")]
|
|
{
|
|
// 检查内核版本和io_uring支持
|
|
if let Ok(uname) = std::process::Command::new("uname").arg("-r").output() {
|
|
let kernel_version = String::from_utf8_lossy(&uname.stdout);
|
|
tracing::info!(target: "sol_trade_sdk","Kernel version: {}", kernel_version.trim());
|
|
|
|
// 简单检查:内核版本 >= 5.1 支持io_uring
|
|
if let Some(version_str) = kernel_version.split('.').next() {
|
|
if let Ok(major_version) = version_str.parse::<u32>() {
|
|
return major_version >= 5;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
false
|
|
}
|
|
|
|
/// 🚀 批量异步写入 - 绕过多次系统调用
|
|
#[inline(always)]
|
|
pub async fn batch_async_write(&self, requests: &[(i32, &[u8])]) -> Result<Vec<usize>> {
|
|
if self.io_uring_available && requests.len() > 1 {
|
|
return self.io_uring_batch_write(requests).await;
|
|
}
|
|
|
|
// 回退到标准批量写入
|
|
self.standard_batch_write(requests).await
|
|
}
|
|
|
|
/// 使用io_uring进行批量写入
|
|
async fn io_uring_batch_write(&self, requests: &[(i32, &[u8])]) -> Result<Vec<usize>> {
|
|
// 这里是伪代码 - 实际实现需要io_uring库
|
|
tracing::trace!(target: "sol_trade_sdk","Using io_uring for {} write operations", requests.len());
|
|
|
|
let mut results = Vec::with_capacity(requests.len());
|
|
|
|
// 模拟批量提交到io_uring
|
|
for (_fd, data) in requests {
|
|
self.async_io_stats.operations_queued.fetch_add(1, Ordering::Relaxed);
|
|
|
|
// 实际的io_uring实现会在这里提交所有操作
|
|
// 然后等待完成,避免多次系统调用
|
|
|
|
results.push(data.len()); // 模拟写入成功
|
|
self.async_io_stats.bytes_transferred.fetch_add(data.len() as u64, Ordering::Relaxed);
|
|
self.async_io_stats.operations_completed.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
|
|
// 这是一个系统调用而不是N个
|
|
self.async_io_stats.syscalls_avoided.fetch_add(requests.len() as u64 - 1, Ordering::Relaxed);
|
|
|
|
Ok(results)
|
|
}
|
|
|
|
/// 标准批量写入
|
|
async fn standard_batch_write(&self, requests: &[(i32, &[u8])]) -> Result<Vec<usize>> {
|
|
let mut results = Vec::with_capacity(requests.len());
|
|
|
|
// 将所有写入打包成一个写操作
|
|
for (_fd, data) in requests {
|
|
// 模拟写入操作
|
|
results.push(data.len());
|
|
self.async_io_stats.bytes_transferred.fetch_add(data.len() as u64, Ordering::Relaxed);
|
|
}
|
|
|
|
Ok(results)
|
|
}
|
|
|
|
/// 🚀 内存映射文件I/O - 避免read/write系统调用
|
|
pub fn create_memory_mapped_io(&mut self, file_path: &str, size: usize) -> Result<usize> {
|
|
#[cfg(unix)]
|
|
{
|
|
use std::fs::OpenOptions;
|
|
use std::os::fd::AsRawFd;
|
|
|
|
#[cfg(target_os = "linux")]
|
|
let file = {
|
|
use std::os::unix::fs::OpenOptionsExt;
|
|
OpenOptions::new()
|
|
.read(true)
|
|
.write(true)
|
|
.create(true)
|
|
.custom_flags(libc::O_DIRECT) // 直接I/O,绕过页面缓存
|
|
.open(file_path)?
|
|
};
|
|
|
|
#[cfg(not(target_os = "linux"))]
|
|
let file = OpenOptions::new()
|
|
.read(true)
|
|
.write(true)
|
|
.create(true)
|
|
.open(file_path)?;
|
|
|
|
let fd = file.as_raw_fd();
|
|
|
|
unsafe {
|
|
let addr = libc::mmap(
|
|
std::ptr::null_mut(),
|
|
size,
|
|
libc::PROT_READ | libc::PROT_WRITE,
|
|
libc::MAP_SHARED,
|
|
fd,
|
|
0,
|
|
);
|
|
|
|
if addr == libc::MAP_FAILED {
|
|
return Err(anyhow::anyhow!("Memory mapping failed"));
|
|
}
|
|
|
|
let region = MemoryMappedRegion {
|
|
address: addr as usize,
|
|
size,
|
|
file_descriptor: fd,
|
|
};
|
|
|
|
self.mmap_regions.push(region);
|
|
|
|
tracing::info!(target: "sol_trade_sdk","✅ Memory mapped I/O created: {} bytes at {:p}", size, addr);
|
|
Ok(addr as usize)
|
|
}
|
|
}
|
|
|
|
#[cfg(not(unix))]
|
|
{
|
|
Err(anyhow::anyhow!("Memory mapped I/O not supported on this platform"))
|
|
}
|
|
}
|
|
|
|
/// 获取I/O统计
|
|
pub fn get_stats(&self) -> AsyncIOStats {
|
|
AsyncIOStats {
|
|
operations_queued: AtomicU64::new(self.async_io_stats.operations_queued.load(Ordering::Relaxed)),
|
|
operations_completed: AtomicU64::new(self.async_io_stats.operations_completed.load(Ordering::Relaxed)),
|
|
bytes_transferred: AtomicU64::new(self.async_io_stats.bytes_transferred.load(Ordering::Relaxed)),
|
|
syscalls_avoided: AtomicU64::new(self.async_io_stats.syscalls_avoided.load(Ordering::Relaxed)),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl SyscallBatchProcessor {
|
|
/// 创建系统调用批处理器
|
|
pub fn new(batch_size: usize) -> Result<Self> {
|
|
let pending_calls = crossbeam_queue::ArrayQueue::new(batch_size * 10);
|
|
let executor = tokio::runtime::Handle::current();
|
|
|
|
tracing::info!(target: "sol_trade_sdk","🚀 Syscall batch processor created with batch size: {}", batch_size);
|
|
|
|
Ok(Self {
|
|
pending_calls,
|
|
_executor: executor,
|
|
batch_stats: CachePadded::new(AtomicU64::new(0)),
|
|
})
|
|
}
|
|
|
|
/// 🚀 提交系统调用请求到批处理队列
|
|
#[inline(always)]
|
|
pub fn submit_request(&self, request: SyscallRequest) -> Result<()> {
|
|
self.pending_calls.push(request)
|
|
.map_err(|_| anyhow::anyhow!("Batch queue full"))?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// 🚀 执行批量系统调用
|
|
pub async fn execute_batch(&self) -> Result<usize> {
|
|
let mut batch = Vec::new();
|
|
|
|
// 收集批量请求
|
|
while batch.len() < 100 && !self.pending_calls.is_empty() {
|
|
if let Some(request) = self.pending_calls.pop() {
|
|
batch.push(request);
|
|
}
|
|
}
|
|
|
|
if batch.is_empty() {
|
|
return Ok(0);
|
|
}
|
|
|
|
let batch_size = batch.len();
|
|
|
|
// 按类型分组批量执行
|
|
let mut write_requests = Vec::new();
|
|
let mut read_requests = Vec::new();
|
|
let mut network_requests = Vec::new();
|
|
|
|
for request in batch {
|
|
match request {
|
|
SyscallRequest::Write { fd, data } => {
|
|
write_requests.push((fd, data));
|
|
}
|
|
SyscallRequest::Read { fd, size } => {
|
|
read_requests.push((fd, size));
|
|
}
|
|
SyscallRequest::Send { socket, data } => {
|
|
network_requests.push((socket, data));
|
|
}
|
|
_ => {
|
|
// 其他类型的请求单独处理
|
|
}
|
|
}
|
|
}
|
|
|
|
// 批量执行写入
|
|
if !write_requests.is_empty() {
|
|
self.batch_write_operations(write_requests).await?;
|
|
}
|
|
|
|
// 批量执行读取
|
|
if !read_requests.is_empty() {
|
|
self.batch_read_operations(read_requests).await?;
|
|
}
|
|
|
|
// 批量执行网络操作
|
|
if !network_requests.is_empty() {
|
|
self.batch_network_operations(network_requests).await?;
|
|
}
|
|
|
|
self.batch_stats.fetch_add(1, Ordering::Relaxed);
|
|
|
|
tracing::trace!(target: "sol_trade_sdk","Executed batch of {} syscalls", batch_size);
|
|
Ok(batch_size)
|
|
}
|
|
|
|
/// 批量写入操作
|
|
async fn batch_write_operations(&self, requests: Vec<(i32, Vec<u8>)>) -> Result<()> {
|
|
// 使用writev系统调用进行批量写入
|
|
for (fd, data) in requests {
|
|
// 实际实现会使用writev或io_uring
|
|
tracing::trace!(target: "sol_trade_sdk","Batched write to fd {}: {} bytes", fd, data.len());
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// 批量读取操作
|
|
async fn batch_read_operations(&self, requests: Vec<(i32, usize)>) -> Result<()> {
|
|
// 使用readv系统调用进行批量读取
|
|
for (fd, size) in requests {
|
|
tracing::trace!(target: "sol_trade_sdk","Batched read from fd {}: {} bytes", fd, size);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// 批量网络操作
|
|
async fn batch_network_operations(&self, requests: Vec<(i32, Vec<u8>)>) -> Result<()> {
|
|
// 使用sendmsg/recvmsg进行批量网络操作
|
|
for (socket, data) in requests {
|
|
tracing::trace!(target: "sol_trade_sdk","Batched network send to socket {}: {} bytes", socket, data.len());
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// 系统调用绕过统计
|
|
#[derive(Debug, Default)]
|
|
pub struct SyscallBypassStats {
|
|
pub syscalls_bypassed: AtomicU64,
|
|
pub syscalls_batched: AtomicU64,
|
|
pub time_calls_cached: AtomicU64,
|
|
pub io_operations_optimized: AtomicU64,
|
|
pub memory_operations_avoided: AtomicU64,
|
|
}
|
|
|
|
impl SystemCallBypassManager {
|
|
/// 创建系统调用绕过管理器
|
|
pub fn new(config: SyscallBypassConfig) -> Result<Self> {
|
|
let batch_processor = Arc::new(SyscallBatchProcessor::new(config.batch_size)?);
|
|
let fast_time_provider = Arc::new(FastTimeProvider::new(config.enable_vdso)?);
|
|
let io_optimizer = Arc::new(IOOptimizer::new(&config)?);
|
|
let stats = Arc::new(SyscallBypassStats::default());
|
|
|
|
tracing::info!(target: "sol_trade_sdk","🚀 System Call Bypass Manager initialized");
|
|
tracing::info!(target: "sol_trade_sdk"," 📦 Batch Processing: {}", config.enable_batch_processing);
|
|
tracing::info!(target: "sol_trade_sdk"," ⏰ Fast Time: {}", config.enable_fast_time);
|
|
tracing::info!(target: "sol_trade_sdk"," 🚀 vDSO: {}", config.enable_vdso);
|
|
tracing::info!(target: "sol_trade_sdk"," 📁 io_uring: {}", config.enable_io_uring);
|
|
|
|
Ok(Self {
|
|
config,
|
|
batch_processor,
|
|
fast_time_provider,
|
|
_io_optimizer: io_optimizer,
|
|
stats,
|
|
})
|
|
}
|
|
|
|
/// 🚀 快速获取当前时间戳 - 绕过系统调用
|
|
#[inline(always)]
|
|
pub fn fast_timestamp_nanos(&self) -> u64 {
|
|
if self.config.enable_fast_time {
|
|
self.stats.time_calls_cached.fetch_add(1, Ordering::Relaxed);
|
|
return self.fast_time_provider.fast_now_nanos();
|
|
}
|
|
|
|
// 回退到标准时间获取
|
|
SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap_or_default()
|
|
.as_nanos() as u64
|
|
}
|
|
|
|
/// 🚀 提交批量I/O操作
|
|
pub async fn submit_batch_io(&self, operations: Vec<SyscallRequest>) -> Result<()> {
|
|
if !self.config.enable_batch_processing {
|
|
return Err(anyhow::anyhow!("Batch processing disabled"));
|
|
}
|
|
|
|
for op in operations {
|
|
self.batch_processor.submit_request(op)?;
|
|
}
|
|
|
|
self.stats.syscalls_batched.fetch_add(1, Ordering::Relaxed);
|
|
Ok(())
|
|
}
|
|
|
|
/// 🚀 执行优化的内存分配 - 绕过malloc系统调用
|
|
#[inline(always)]
|
|
pub fn fast_allocate(&self, size: usize) -> Result<*mut u8> {
|
|
if self.config.enable_userspace_impl {
|
|
self.stats.memory_operations_avoided.fetch_add(1, Ordering::Relaxed);
|
|
return self.userspace_allocate(size);
|
|
}
|
|
|
|
// 回退到标准分配
|
|
let layout = std::alloc::Layout::from_size_align(size, 8)?;
|
|
let ptr = unsafe { std::alloc::alloc(layout) };
|
|
|
|
if ptr.is_null() {
|
|
Err(anyhow::anyhow!("Allocation failed"))
|
|
} else {
|
|
Ok(ptr)
|
|
}
|
|
}
|
|
|
|
/// 用户空间内存分配
|
|
fn userspace_allocate(&self, size: usize) -> Result<*mut u8> {
|
|
use std::sync::Mutex;
|
|
use once_cell::sync::Lazy;
|
|
|
|
struct MemoryPool {
|
|
pool: Box<[u8; 1024 * 1024]>,
|
|
offset: usize,
|
|
}
|
|
|
|
static MEMORY_POOL: Lazy<Mutex<MemoryPool>> = Lazy::new(|| {
|
|
Mutex::new(MemoryPool {
|
|
pool: Box::new([0; 1024 * 1024]),
|
|
offset: 0,
|
|
})
|
|
});
|
|
|
|
let mut pool = MEMORY_POOL.lock().unwrap();
|
|
|
|
if pool.offset + size > pool.pool.len() {
|
|
return Err(anyhow::anyhow!("Memory pool exhausted"));
|
|
}
|
|
|
|
let ptr = unsafe { pool.pool.as_mut_ptr().add(pool.offset) };
|
|
pool.offset += (size + 7) & !7; // 8字节对齐
|
|
|
|
Ok(ptr)
|
|
}
|
|
|
|
/// 启动批处理工作线程
|
|
pub async fn start_batch_processing(&self) -> Result<()> {
|
|
let processor = Arc::clone(&self.batch_processor);
|
|
let stats = Arc::clone(&self.stats);
|
|
|
|
tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(Duration::from_micros(100)); // 100μs间隔
|
|
|
|
loop {
|
|
interval.tick().await;
|
|
|
|
if let Ok(processed) = processor.execute_batch().await {
|
|
if processed > 0 {
|
|
stats.syscalls_bypassed.fetch_add(processed as u64, Ordering::Relaxed);
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
tracing::info!(target: "sol_trade_sdk","✅ Batch processing worker started");
|
|
Ok(())
|
|
}
|
|
|
|
/// 获取绕过统计
|
|
pub fn get_bypass_stats(&self) -> SyscallBypassStatsSnapshot {
|
|
SyscallBypassStatsSnapshot {
|
|
syscalls_bypassed: self.stats.syscalls_bypassed.load(Ordering::Relaxed),
|
|
syscalls_batched: self.stats.syscalls_batched.load(Ordering::Relaxed),
|
|
time_calls_cached: self.stats.time_calls_cached.load(Ordering::Relaxed),
|
|
io_operations_optimized: self.stats.io_operations_optimized.load(Ordering::Relaxed),
|
|
memory_operations_avoided: self.stats.memory_operations_avoided.load(Ordering::Relaxed),
|
|
}
|
|
}
|
|
|
|
/// 🚀 极致优化配置
|
|
pub fn extreme_bypass_config() -> SyscallBypassConfig {
|
|
SyscallBypassConfig {
|
|
enable_batch_processing: true,
|
|
batch_size: 1000, // 更大的批量
|
|
enable_fast_time: true,
|
|
enable_vdso: true,
|
|
enable_io_uring: true,
|
|
enable_mmap_optimization: true,
|
|
enable_userspace_impl: true,
|
|
syscall_cache_size: 10000,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 系统调用绕过统计快照
|
|
#[derive(Debug, Clone)]
|
|
pub struct SyscallBypassStatsSnapshot {
|
|
pub syscalls_bypassed: u64,
|
|
pub syscalls_batched: u64,
|
|
pub time_calls_cached: u64,
|
|
pub io_operations_optimized: u64,
|
|
pub memory_operations_avoided: u64,
|
|
}
|
|
|
|
impl SyscallBypassStatsSnapshot {
|
|
/// 打印统计信息
|
|
pub fn print_stats(&self) {
|
|
tracing::info!(target: "sol_trade_sdk","📊 System Call Bypass Stats:");
|
|
tracing::info!(target: "sol_trade_sdk"," 🚫 Syscalls Bypassed: {}", self.syscalls_bypassed);
|
|
tracing::info!(target: "sol_trade_sdk"," 📦 Syscalls Batched: {}", self.syscalls_batched);
|
|
tracing::info!(target: "sol_trade_sdk"," ⏰ Time Calls Cached: {}", self.time_calls_cached);
|
|
tracing::info!(target: "sol_trade_sdk"," 📁 I/O Operations Optimized: {}", self.io_operations_optimized);
|
|
tracing::info!(target: "sol_trade_sdk"," 💾 Memory Operations Avoided: {}", self.memory_operations_avoided);
|
|
|
|
let total_optimizations = self.syscalls_bypassed + self.time_calls_cached +
|
|
self.io_operations_optimized + self.memory_operations_avoided;
|
|
tracing::info!(target: "sol_trade_sdk"," 🏆 Total Optimizations: {}", total_optimizations);
|
|
}
|
|
}
|
|
|
|
/// 🚀 系统调用绕过宏
|
|
#[macro_export]
|
|
macro_rules! bypass_syscall {
|
|
(time) => {
|
|
// 使用快速时间而不是系统调用
|
|
crate::performance::syscall_bypass::GLOBAL_TIME_PROVIDER.fast_now_nanos()
|
|
};
|
|
|
|
(batch_io $ops:expr) => {
|
|
// 批量提交I/O操作
|
|
crate::performance::syscall_bypass::GLOBAL_BYPASS_MANAGER.submit_batch_io($ops).await
|
|
};
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn test_fast_time_provider() {
|
|
let provider = FastTimeProvider::new(false).unwrap();
|
|
|
|
let time1 = provider.fast_now_nanos();
|
|
tokio::time::sleep(Duration::from_millis(1)).await;
|
|
let time2 = provider.fast_now_nanos();
|
|
|
|
assert!(time2 > time1);
|
|
assert!(time2 - time1 >= 1_000_000); // 至少1ms差异
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_syscall_batch_processor() {
|
|
let processor = SyscallBatchProcessor::new(10).unwrap();
|
|
|
|
let request = SyscallRequest::Write {
|
|
fd: 1,
|
|
data: vec![1, 2, 3, 4, 5],
|
|
};
|
|
|
|
processor.submit_request(request).unwrap();
|
|
|
|
let processed = processor.execute_batch().await.unwrap();
|
|
assert_eq!(processed, 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_io_optimizer() {
|
|
let config = SyscallBypassConfig::default();
|
|
let optimizer = IOOptimizer::new(&config).unwrap();
|
|
|
|
let requests = vec![(1, b"test data".as_ref())];
|
|
let results = optimizer.batch_async_write(&requests).await.unwrap();
|
|
|
|
assert_eq!(results.len(), 1);
|
|
assert_eq!(results[0], 9); // "test data".len()
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_system_call_bypass_manager() {
|
|
let config = SyscallBypassConfig::default();
|
|
let manager = SystemCallBypassManager::new(config).unwrap();
|
|
|
|
// 测试快速时间戳
|
|
let timestamp = manager.fast_timestamp_nanos();
|
|
assert!(timestamp > 0);
|
|
|
|
// 测试统计
|
|
let stats = manager.get_bypass_stats();
|
|
assert_eq!(stats.time_calls_cached, 1);
|
|
}
|
|
|
|
#[test]
|
|
fn test_extreme_bypass_config() {
|
|
let config = SystemCallBypassManager::extreme_bypass_config();
|
|
assert!(config.enable_batch_processing);
|
|
assert!(config.enable_fast_time);
|
|
assert!(config.enable_vdso);
|
|
assert!(config.enable_io_uring);
|
|
assert_eq!(config.batch_size, 1000);
|
|
assert_eq!(config.syscall_cache_size, 10000);
|
|
}
|
|
|
|
#[test]
|
|
fn test_userspace_allocation() {
|
|
let config = SyscallBypassConfig::default();
|
|
let manager = SystemCallBypassManager::new(config).unwrap();
|
|
|
|
let ptr = manager.fast_allocate(64).unwrap();
|
|
assert!(!ptr.is_null());
|
|
|
|
let stats = manager.get_bypass_stats();
|
|
assert_eq!(stats.memory_operations_avoided, 1);
|
|
}
|
|
} |