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

style: run rustfmt

Browse Source
This commit is contained in:
hookenful
2026-03-14 18:16:55 +02:00
parent 0ec826fcbd
commit 2abcf3839e
76 changed files with 1763 additions and 1352 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/perf/compiler_optimization.rs
+90 -75
View File
@@ -1,5 +1,5 @@
//! 🚀 编译器级性能优化 - 极致编译时优化
//!
//!
//! 实现编译时的极致性能优化，包括：
//! - 编译器标志优化配置
//! - 编译时代码生成
@@ -137,100 +137,110 @@ impl CompilerOptimizer {
stats: CompilerOptimizationStats::default(),
}
}
/// 🚀 生成超高性能编译配置
pub fn generate_ultra_performance_config(&self) -> Result<CompilerConfig> {
tracing::info!(target: "sol_trade_sdk","🚀 Generating ultra-performance compiler configuration...");
let mut rustflags = Vec::new();
// 基础优化标志
rustflags.push("-C".to_string());
rustflags.push("opt-level=3".to_string()); // 最高优化级别
// 链接时优化
if self.optimization_flags.enable_lto {
rustflags.push("-C".to_string());
rustflags.push("lto=fat".to_string()); // 胖LTO获得最佳优化
}
// 目标CPU优化
if !self.optimization_flags.target_cpu.is_empty() {
rustflags.push("-C".to_string());
rustflags.push(format!("target-cpu={}", self.optimization_flags.target_cpu));
}
// 目标特性
if !self.optimization_flags.target_features.is_empty() {
rustflags.push("-C".to_string());
rustflags.push(format!("target-feature={}", self.optimization_flags.target_features.join(",")));
rustflags.push(format!(
"target-feature={}",
self.optimization_flags.target_features.join(",")
));
}
// 代码模型
rustflags.push("-C".to_string());
rustflags.push(format!("code-model={:?}", self.optimization_flags.code_model).to_lowercase());
rustflags
.push(format!("code-model={:?}", self.optimization_flags.code_model).to_lowercase());
// 恐慌处理
if self.codegen_config.panic_abort {
rustflags.push("-C".to_string());
rustflags.push("panic=abort".to_string());
}
// 溢出检查
if !self.codegen_config.overflow_checks {
rustflags.push("-C".to_string());
rustflags.push("overflow-checks=no".to_string());
}
// 代码生成单元
if let Some(units) = self.optimization_flags.codegen_units {
rustflags.push("-C".to_string());
rustflags.push(format!("codegen-units={}", units));
}
// 内联阈值
rustflags.push("-C".to_string());
rustflags.push(format!("inline-threshold={}", self.inline_strategy.inline_threshold));
// 额外的性能优化标志
rustflags.extend([
"-C".to_string(), "embed-bitcode=no".to_string(), // 不嵌入位码以减少体积
"-C".to_string(), "debuginfo=0".to_string(), // 禁用调试信息
"-C".to_string(), "rpath=no".to_string(), // 禁用rpath
"-C".to_string(), "force-frame-pointers=no".to_string(), // 禁用帧指针
"-C".to_string(),
"embed-bitcode=no".to_string(), // 不嵌入位码以减少体积
"-C".to_string(),
"debuginfo=0".to_string(), // 禁用调试信息
"-C".to_string(),
"rpath=no".to_string(), // 禁用rpath
"-C".to_string(),
"force-frame-pointers=no".to_string(), // 禁用帧指针
]);
let config = CompilerConfig {
rustflags,
env_vars: self.generate_env_vars(),
cargo_config: self.generate_cargo_config(),
};
tracing::info!(target: "sol_trade_sdk","✅ Ultra-performance compiler configuration generated");
Ok(config)
}
/// 生成环境变量配置
fn generate_env_vars(&self) -> HashMap<String, String> {
let mut env_vars = HashMap::new();
// CPU特定优化
env_vars.insert("CARGO_CFG_TARGET_FEATURE".to_string(),
self.optimization_flags.target_features.join(","));
env_vars.insert(
"CARGO_CFG_TARGET_FEATURE".to_string(),
self.optimization_flags.target_features.join(","),
);
// 启用不稳定特性
env_vars.insert("RUSTC_BOOTSTRAP".to_string(), "1".to_string());
// 编译缓存设置
if self.optimization_flags.incremental {
env_vars.insert("CARGO_INCREMENTAL".to_string(), "1".to_string());
} else {
env_vars.insert("CARGO_INCREMENTAL".to_string(), "0".to_string());
}
env_vars
}
/// 生成Cargo配置
fn generate_cargo_config(&self) -> CargoConfig {
CargoConfig {
@@ -244,17 +254,21 @@ impl CompilerOptimizer {
debug_assertions: false,
rpath: false,
strip: true, // 去除符号表
}
},
}
}
/// 获取统计信息
pub fn get_stats(&self) -> CompilerOptimizationStats {
CompilerOptimizationStats {
inlined_functions: AtomicU64::new(self.stats.inlined_functions.load(Ordering::Relaxed)),
constant_folding: AtomicU64::new(self.stats.constant_folding.load(Ordering::Relaxed)),
dead_code_elimination: AtomicU64::new(self.stats.dead_code_elimination.load(Ordering::Relaxed)),
loop_optimizations: AtomicU64::new(self.stats.loop_optimizations.load(Ordering::Relaxed)),
dead_code_elimination: AtomicU64::new(
self.stats.dead_code_elimination.load(Ordering::Relaxed),
),
loop_optimizations: AtomicU64::new(
self.stats.loop_optimizations.load(Ordering::Relaxed),
),
}
}
}
@@ -279,12 +293,12 @@ impl OptimizationFlags {
Self {
opt_level: OptLevel::Aggressive,
enable_lto: true,
enable_pgo: false, // PGO需要多阶段构建
enable_pgo: false, // PGO需要多阶段构建
target_cpu: "native".to_string(), // 使用本机CPU特性
target_features,
code_model: CodeModel::Small,
debug_info: false,
incremental: false, // 发布版本禁用增量编译
incremental: false, // 发布版本禁用增量编译
codegen_units: Some(1), // 单个代码生成单元获得最佳优化
}
}
@@ -294,7 +308,7 @@ impl CodegenConfig {
/// 超高性能配置
pub fn ultra_performance() -> Self {
Self {
panic_abort: true, // 恐慌即中止，避免展开开销
panic_abort: true, // 恐慌即中止，避免展开开销
overflow_checks: false, // 生产环境禁用溢出检查
fat_lto: true,
enable_simd: true,
@@ -353,14 +367,14 @@ macro_rules! compile_time_optimize {
(const $expr:expr) => {
const { $expr }
};
// 强制内联热路径
(inline_hot $fn_name:ident) => {
#[inline(always)]
#[hot]
$fn_name
};
// 标记冷路径
(cold $fn_name:ident) => {
#[inline(never)]
@@ -372,10 +386,10 @@ macro_rules! compile_time_optimize {
/// 🚀 零成本抽象特征
pub trait ZeroCostAbstraction {
type Output;
/// 编译时计算
fn compute_at_compile_time(&self) -> Self::Output;
/// 内联操作
#[inline(always)]
fn inline_operation(&self) -> Self::Output {
@@ -399,35 +413,35 @@ impl CompileTimeOptimizedEventProcessor {
route_table: Self::precompute_route_table(),
}
}
/// 编译时预计算哈希表
const fn precompute_hash_table() -> [u64; 256] {
let mut table = [0u64; 256];
let mut i = 0;
while i < 256 {
// 使用编译时常量计算哈希值
table[i] = Self::const_hash(i as u8);
i += 1;
}
table
}
/// 编译时预计算路由表
const fn precompute_route_table() -> [u32; 1024] {
let mut table = [0u32; 1024];
let mut i = 0;
while i < 1024 {
// 预计算路由信息
table[i] = (i as u32) % 16; // 16个工作线程
i += 1;
}
table
}
/// 编译时常量哈希函数
const fn const_hash(input: u8) -> u64 {
// 使用简单的编译时常量哈希
@@ -437,16 +451,14 @@ impl CompileTimeOptimizedEventProcessor {
hash ^= hash << 17;
hash
}
/// 🚀 零开销事件路由
#[inline(always)]
pub fn route_event_zero_cost(&self, event_id: u8) -> u32 {
// 编译时优化：直接数组访问，无边界检查
unsafe {
*self.route_table.get_unchecked((event_id as usize) & 1023)
}
unsafe { *self.route_table.get_unchecked((event_id as usize) & 1023) }
}
/// 🚀 编译时优化的哈希查找
#[inline(always)]
pub fn hash_lookup_optimized(&self, key: u8) -> u64 {
@@ -464,28 +476,28 @@ impl SIMDCompileTimeOptimizer {
#[target_feature(enable = "avx2")]
pub unsafe fn vectorized_sum_compile_time(data: &[u64]) -> u64 {
use std::arch::x86_64::*;
if data.len() < 4 {
return data.iter().sum();
}
let chunks = data.len() / 4;
let mut sum_vec = _mm256_setzero_si256();
for i in 0..chunks {
let ptr = data.as_ptr().add(i * 4) as *const __m256i;
let vec = _mm256_loadu_si256(ptr);
sum_vec = _mm256_add_epi64(sum_vec, vec);
}
// 水平求和
let mut result = [0u64; 4];
_mm256_storeu_si256(result.as_mut_ptr() as *mut __m256i, sum_vec);
let partial_sum: u64 = result.iter().sum();
// 处理剩余元素
let remaining: u64 = data[chunks * 4..].iter().sum();
partial_sum + remaining
}
@@ -523,7 +535,8 @@ fn main() {
println!("cargo:rustc-link-arg=-fprofile-use");
}
}
"#.to_string()
"#
.to_string()
}
/// 🚀 生成.cargo/config.toml
@@ -576,56 +589,58 @@ rustflags = [
rustflags = [
"-C", "target-feature=+sse4.2,+avx,+avx2,+fma,+bmi1,+bmi2,+lzcnt,+popcnt",
]
"#.to_string()
"#
.to_string()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_compiler_optimizer_creation() {
let optimizer = CompilerOptimizer::new();
assert!(optimizer.optimization_flags.enable_lto);
assert_eq!(optimizer.optimization_flags.opt_level as u8, OptLevel::Aggressive as u8);
}
#[test]
fn test_compile_time_processor() {
const PROCESSOR: CompileTimeOptimizedEventProcessor = CompileTimeOptimizedEventProcessor::new();
const PROCESSOR: CompileTimeOptimizedEventProcessor =
CompileTimeOptimizedEventProcessor::new();
let route = PROCESSOR.route_event_zero_cost(42);
assert!(route < 16); // 应该路由到16个工作线程之一
let hash = PROCESSOR.hash_lookup_optimized(100);
assert!(hash > 0); // 哈希值应该非零
}
#[test]
fn test_ultra_performance_config() {
let flags = OptimizationFlags::ultra_performance();
assert!(flags.enable_lto);
assert_eq!(flags.target_cpu, "native");
assert!(!flags.target_features.is_empty());
let codegen = CodegenConfig::ultra_performance();
assert!(codegen.panic_abort);
assert!(!codegen.overflow_checks);
assert!(codegen.enable_simd);
}
#[test]
#[test]
fn test_compiler_config_generation() {
let optimizer = CompilerOptimizer::new();
let config = optimizer.generate_ultra_performance_config().unwrap();
assert!(!config.rustflags.is_empty());
assert!(config.rustflags.contains(&"-C".to_string()));
assert!(config.rustflags.contains(&"opt-level=3".to_string()));
assert!(config.env_vars.contains_key("CARGO_INCREMENTAL"));
}
#[test]
fn test_simd_compile_time_optimization() {
#[cfg(any(target_arch = "x86", target_arch = "x86_64"))]
@@ -642,7 +657,7 @@ mod tests {
assert_eq!(sum, 36); // 1+2+3+4+5+6+7+8 = 36
}
}
#[test]
fn test_build_script_generation() {
let build_script = generate_build_script();
@@ -650,7 +665,7 @@ mod tests {
assert!(build_script.contains("TARGET_FEATURE"));
assert!(build_script.contains("lld"));
}
#[test]
fn test_cargo_config_generation() {
let config = generate_cargo_config_toml();
@@ -659,4 +674,4 @@ mod tests {
assert!(config.contains("target-cpu=native"));
assert!(config.contains("panic = \"abort\""));
}
}
}
src/perf/hardware_optimizations.rs
+82 -93
View File
@@ -1,11 +1,11 @@
//! Hardware-oriented optimizations: cache-line alignment, prefetch, SIMD, branch hints, memory barriers.
//! 硬件级优化：缓存行对齐与预取、SIMD、分支提示、内存屏障。
use std::sync::atomic::{AtomicU64, Ordering};
use anyhow::Result;
use crossbeam_utils::CachePadded;
use std::mem::size_of;
use std::ptr;
use crossbeam_utils::CachePadded;
use anyhow::Result;
use std::sync::atomic::{AtomicU64, Ordering};
/// Typical CPU cache line size in bytes. 典型 CPU 缓存行大小（字节）。
pub const CACHE_LINE_SIZE: usize = 64;
@@ -32,7 +32,7 @@ impl SIMDMemoryOps {
_ => Self::memcpy_avx512_or_fallback(dst, src, len),
}
}
/// Copy 1–8 bytes (scalar / small word). 小数据拷贝（1–8 字节）。
#[inline(always)]
unsafe fn memcpy_small(dst: *mut u8, src: *const u8, len: usize) {
@@ -53,52 +53,52 @@ impl SIMDMemoryOps {
_ => unreachable!(),
}
}
/// Copy 9–16 bytes using SSE (128-bit). SSE 拷贝（9–16 字节）。
#[inline(always)]
unsafe fn memcpy_sse(dst: *mut u8, src: *const u8, len: usize) {
#[cfg(target_arch = "x86_64")]
{
use std::arch::x86_64::{__m128i, _mm_loadu_si128, _mm_storeu_si128};
if len <= 16 {
let chunk = _mm_loadu_si128(src as *const __m128i);
_mm_storeu_si128(dst as *mut __m128i, chunk);
}
}
#[cfg(not(target_arch = "x86_64"))]
{
ptr::copy_nonoverlapping(src, dst, len);
}
}
/// Copy 17–32 bytes using AVX (256-bit). AVX 拷贝（17–32 字节）。
#[inline(always)]
unsafe fn memcpy_avx(dst: *mut u8, src: *const u8, len: usize) {
#[cfg(target_arch = "x86_64")]
{
use std::arch::x86_64::{__m256i, _mm256_loadu_si256, _mm256_storeu_si256};
if len <= 32 {
let chunk = _mm256_loadu_si256(src as *const __m256i);
_mm256_storeu_si256(dst as *mut __m256i, chunk);
}
}
#[cfg(not(target_arch = "x86_64"))]
{
ptr::copy_nonoverlapping(src, dst, len);
}
}
/// Copy 33–64 bytes using AVX2 (256-bit, two chunks). AVX2 拷贝（33–64 字节，两段）。
#[inline(always)]
unsafe fn memcpy_avx2(dst: *mut u8, src: *const u8, len: usize) {
#[cfg(target_arch = "x86_64")]
{
use std::arch::x86_64::{__m256i, _mm256_loadu_si256, _mm256_storeu_si256};
let chunk1 = _mm256_loadu_si256(src as *const __m256i);
_mm256_storeu_si256(dst as *mut __m256i, chunk1);
if len > 32 {
@@ -109,52 +109,52 @@ impl SIMDMemoryOps {
}
}
}
#[cfg(not(target_arch = "x86_64"))]
{
ptr::copy_nonoverlapping(src, dst, len);
}
}
/// Copy >64 bytes: AVX-512 64-byte chunks when available, else AVX2 32-byte chunks. >64 字节：有 AVX512 用 64 字节块，否则 AVX2 32 字节块。
#[inline(always)]
unsafe fn memcpy_avx512_or_fallback(dst: *mut u8, src: *const u8, len: usize) {
#[cfg(all(target_arch = "x86_64", target_feature = "avx512f"))]
{
use std::arch::x86_64::{__m512i, _mm512_loadu_si512, _mm512_storeu_si512};
let chunks = len / 64;
let mut offset = 0;
for _ in 0..chunks {
let chunk = _mm512_loadu_si512(src.add(offset) as *const __m512i);
_mm512_storeu_si512(dst.add(offset) as *mut __m512i, chunk);
offset += 64;
}
let remaining = len % 64;
if remaining > 0 {
Self::memcpy_avx2(dst.add(offset), src.add(offset), remaining);
}
}
#[cfg(not(all(target_arch = "x86_64", target_feature = "avx512f")))]
{
let chunks = len / 32;
let mut offset = 0;
for _ in 0..chunks {
Self::memcpy_avx2(dst.add(offset), src.add(offset), 32);
offset += 32;
}
let remaining = len % 32;
if remaining > 0 {
Self::memcpy_avx(dst.add(offset), src.add(offset), remaining);
}
}
}
/// SIMD-optimized byte equality; dispatches by length (small / SSE / AVX2 / large). SIMD 加速的内存比较，按长度分派。
#[inline(always)]
pub unsafe fn memcmp_simd_optimized(a: *const u8, b: *const u8, len: usize) -> bool {
@@ -166,109 +166,108 @@ impl SIMDMemoryOps {
_ => Self::memcmp_large(a, b, len),
}
}
/// Compare 1–8 bytes (scalar). 小数据比较（1–8 字节）。
#[inline(always)]
unsafe fn memcmp_small(a: *const u8, b: *const u8, len: usize) -> bool {
match len {
1 => *a == *b,
2 => *(a as *const u16) == *(b as *const u16),
3 => {
*(a as *const u16) == *(b as *const u16) &&
*a.add(2) == *b.add(2)
}
3 => *(a as *const u16) == *(b as *const u16) && *a.add(2) == *b.add(2),
4 => *(a as *const u32) == *(b as *const u32),
5..=8 => *(a as *const u64) == *(b as *const u64),
_ => unreachable!(),
}
}
/// Compare 9–16 bytes using SSE. SSE 比较（9–16 字节）。
#[inline(always)]
unsafe fn memcmp_sse(a: *const u8, b: *const u8, len: usize) -> bool {
#[cfg(target_arch = "x86_64")]
{
use std::arch::x86_64::{__m128i, _mm_loadu_si128, _mm_cmpeq_epi8, _mm_movemask_epi8};
use std::arch::x86_64::{__m128i, _mm_cmpeq_epi8, _mm_loadu_si128, _mm_movemask_epi8};
let chunk_a = _mm_loadu_si128(a as *const __m128i);
let chunk_b = _mm_loadu_si128(b as *const __m128i);
let cmp_result = _mm_cmpeq_epi8(chunk_a, chunk_b);
let mask = _mm_movemask_epi8(cmp_result) as u32;
let valid_mask = if len >= 16 { 0xFFFF } else { (1u32 << len) - 1 };
(mask & valid_mask) == valid_mask
}
#[cfg(not(target_arch = "x86_64"))]
{
(0..len).all(|i| *a.add(i) == *b.add(i))
}
}
/// Compare 17–32 bytes using AVX2. AVX2 比较（17–32 字节）。
#[inline(always)]
unsafe fn memcmp_avx2(a: *const u8, b: *const u8, len: usize) -> bool {
#[cfg(target_arch = "x86_64")]
{
use std::arch::x86_64::{__m256i, _mm256_loadu_si256, _mm256_cmpeq_epi8, _mm256_movemask_epi8};
use std::arch::x86_64::{
__m256i, _mm256_cmpeq_epi8, _mm256_loadu_si256, _mm256_movemask_epi8,
};
let chunk_a = _mm256_loadu_si256(a as *const __m256i);
let chunk_b = _mm256_loadu_si256(b as *const __m256i);
let cmp_result = _mm256_cmpeq_epi8(chunk_a, chunk_b);
let mask = _mm256_movemask_epi8(cmp_result) as u32;
let valid_mask = if len >= 32 { 0xFFFFFFFF } else { (1u32 << len) - 1 };
(mask & valid_mask) == valid_mask
}
#[cfg(not(target_arch = "x86_64"))]
{
(0..len).all(|i| *a.add(i) == *b.add(i))
}
}
/// Compare >32 bytes in 32-byte AVX2 chunks. 大数据比较（32 字节 AVX2 分块）。
#[inline(always)]
unsafe fn memcmp_large(a: *const u8, b: *const u8, len: usize) -> bool {
let chunks = len / 32;
for i in 0..chunks {
let offset = i * 32;
if !Self::memcmp_avx2(a.add(offset), b.add(offset), 32) {
return false;
}
}
let remaining = len % 32;
if remaining > 0 {
return Self::memcmp_avx2(a.add(chunks * 32), b.add(chunks * 32), remaining);
}
true
}
/// SIMD-optimized zero memory. SIMD 加速的内存清零。
#[inline(always)]
pub unsafe fn memzero_simd_optimized(ptr: *mut u8, len: usize) {
#[cfg(target_arch = "x86_64")]
{
use std::arch::x86_64::{__m256i, _mm256_setzero_si256, _mm256_storeu_si256};
let zero = _mm256_setzero_si256();
let chunks = len / 32;
let mut offset = 0;
for _ in 0..chunks {
_mm256_storeu_si256(ptr.add(offset) as *mut __m256i, zero);
offset += 32;
}
let remaining = len % 32;
for i in 0..remaining {
*ptr.add(offset + i) = 0;
}
}
#[cfg(not(target_arch = "x86_64"))]
{
ptr::write_bytes(ptr, 0, len);
@@ -291,17 +290,17 @@ impl CacheAlignedCounter {
_padding: [0; CACHE_LINE_SIZE - size_of::<AtomicU64>()],
}
}
#[inline(always)]
pub fn increment(&self) -> u64 {
self.value.fetch_add(1, Ordering::Relaxed)
}
#[inline(always)]
pub fn load(&self) -> u64 {
self.value.load(Ordering::Relaxed)
}
#[inline(always)]
pub fn store(&self, val: u64) {
self.value.store(val, Ordering::Relaxed)
@@ -312,7 +311,7 @@ impl CacheLineAligned for CacheAlignedCounter {
fn ensure_cache_aligned(&self) -> bool {
(self as *const Self as usize) % CACHE_LINE_SIZE == 0
}
fn prefetch_data(&self) {
#[cfg(target_arch = "x86_64")]
unsafe {
@@ -339,10 +338,10 @@ impl<T: Copy + Default> CacheOptimizedRingBuffer<T> {
if !capacity.is_power_of_two() {
return Err(anyhow::anyhow!("Capacity must be a power of 2"));
}
let mut buffer = Vec::with_capacity(capacity);
buffer.resize_with(capacity, Default::default);
Ok(Self {
buffer,
producer_head: CachePadded::new(AtomicU64::new(0)),
@@ -351,7 +350,7 @@ impl<T: Copy + Default> CacheOptimizedRingBuffer<T> {
mask: capacity - 1,
})
}
/// Lock-free push; returns false if full. 无锁写入，满则返回 false。
#[inline(always)]
pub fn try_push(&self, item: T) -> bool {
@@ -368,7 +367,7 @@ impl<T: Copy + Default> CacheOptimizedRingBuffer<T> {
self.producer_head.store(current_head + 1, Ordering::Release);
true
}
/// Lock-free pop; returns None if empty. 无锁读取，空则返回 None。
#[inline(always)]
pub fn try_pop(&self) -> Option<T> {
@@ -385,7 +384,7 @@ impl<T: Copy + Default> CacheOptimizedRingBuffer<T> {
self.consumer_tail.store(current_tail + 1, Ordering::Release);
Some(item)
}
/// Current number of elements. 当前元素个数。
#[inline(always)]
pub fn len(&self) -> usize {
@@ -393,12 +392,11 @@ impl<T: Copy + Default> CacheOptimizedRingBuffer<T> {
let tail = self.consumer_tail.load(Ordering::Relaxed);
((head + self.capacity as u64 - tail) & self.mask as u64) as usize
}
/// True if no elements. 是否为空。
#[inline(always)]
pub fn is_empty(&self) -> bool {
self.producer_head.load(Ordering::Relaxed) ==
self.consumer_tail.load(Ordering::Relaxed)
self.producer_head.load(Ordering::Relaxed) == self.consumer_tail.load(Ordering::Relaxed)
}
}
@@ -406,7 +404,7 @@ impl<T> CacheLineAligned for CacheOptimizedRingBuffer<T> {
fn ensure_cache_aligned(&self) -> bool {
(self as *const Self as usize) % CACHE_LINE_SIZE == 0
}
fn prefetch_data(&self) {
#[cfg(target_arch = "x86_64")]
unsafe {
@@ -428,25 +426,25 @@ impl BranchOptimizer {
pub fn likely(condition: bool) -> bool {
#[cold]
fn cold() {}
if !condition {
cold();
}
condition
}
/// Hint: condition is usually false. 提示编译器条件大概率为假。
#[inline(always)]
pub fn unlikely(condition: bool) -> bool {
#[cold]
fn cold() {}
if condition {
cold();
}
condition
}
/// Prefetch: load cache line at ptr into L1. Caller must ensure ptr is valid, read-only, no concurrent write. 预取：将 ptr 所在缓存行加载到 L1；调用方需保证有效、只读、无并发写。
#[inline(always)]
pub unsafe fn prefetch_read_data<T>(ptr: *const T) {
@@ -457,7 +455,7 @@ impl BranchOptimizer {
_mm_prefetch(ptr as *const i8, _MM_HINT_T0);
}
}
/// Prefetch for write (T1 hint). 写预取（T1 提示）。
#[inline(always)]
pub unsafe fn prefetch_write_data<T>(ptr: *const T) {
@@ -479,25 +477,25 @@ impl MemoryBarriers {
pub fn compiler_barrier() {
std::sync::atomic::compiler_fence(Ordering::SeqCst);
}
/// Light barrier (Acquire). 轻量级屏障（Acquire）。
#[inline(always)]
pub fn memory_barrier_light() {
std::sync::atomic::fence(Ordering::Acquire);
}
/// Full sequential consistency barrier. 全序一致性屏障。
#[inline(always)]
pub fn memory_barrier_heavy() {
std::sync::atomic::fence(Ordering::SeqCst);
}
/// Store/release barrier. 存储屏障，保证写入可见性。
#[inline(always)]
pub fn store_barrier() {
std::sync::atomic::fence(Ordering::Release);
}
/// Load/acquire barrier. 加载屏障，保证读取顺序。
#[inline(always)]
pub fn load_barrier() {
@@ -508,63 +506,54 @@ impl MemoryBarriers {
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_cache_aligned_counter() {
let counter = CacheAlignedCounter::new(0);
assert!(counter.ensure_cache_aligned());
assert_eq!(counter.load(), 0);
counter.increment();
assert_eq!(counter.load(), 1);
}
#[test]
fn test_simd_memcpy() {
let src = [1u8, 2, 3, 4, 5, 6, 7, 8, 9, 10];
let mut dst = [0u8; 10];
unsafe {
SIMDMemoryOps::memcpy_simd_optimized(
dst.as_mut_ptr(),
src.as_ptr(),
src.len()
);
SIMDMemoryOps::memcpy_simd_optimized(dst.as_mut_ptr(), src.as_ptr(), src.len());
}
assert_eq!(src, dst);
}
#[test]
fn test_cache_optimized_ring_buffer() {
let buffer: CacheOptimizedRingBuffer<u64> =
CacheOptimizedRingBuffer::new(16).unwrap();
let buffer: CacheOptimizedRingBuffer<u64> = CacheOptimizedRingBuffer::new(16).unwrap();
assert!(buffer.is_empty());
// 测试推入
assert!(buffer.try_push(42));
assert_eq!(buffer.len(), 1);
// 测试弹出
assert_eq!(buffer.try_pop(), Some(42));
assert!(buffer.is_empty());
}
#[test]
fn test_simd_memcmp() {
let a = [1u8, 2, 3, 4, 5];
let b = [1u8, 2, 3, 4, 5];
let c = [1u8, 2, 3, 4, 6];
unsafe {
assert!(SIMDMemoryOps::memcmp_simd_optimized(
a.as_ptr(), b.as_ptr(), a.len()
));
assert!(!SIMDMemoryOps::memcmp_simd_optimized(
a.as_ptr(), c.as_ptr(), a.len()
));
assert!(SIMDMemoryOps::memcmp_simd_optimized(a.as_ptr(), b.as_ptr(), a.len()));
assert!(!SIMDMemoryOps::memcmp_simd_optimized(a.as_ptr(), c.as_ptr(), a.len()));
}
}
}
}
src/perf/mod.rs
+8 -8
View File
@@ -1,14 +1,14 @@
//! Performance: SIMD, cache prefetch, branch hints, zero-copy I/O, syscall bypass, compiler hints.
//! 性能优化：SIMD、缓存预取、分支提示、零拷贝 I/O、系统调用绕过、编译器提示。
pub mod simd;
pub mod hardware_optimizations;
pub mod zero_copy_io;
pub mod syscall_bypass;
pub mod compiler_optimization;
pub mod hardware_optimizations;
pub mod simd;
pub mod syscall_bypass;
pub mod zero_copy_io;
pub use simd::*;
pub use hardware_optimizations::*;
pub use zero_copy_io::*;
pub use syscall_bypass::*;
pub use compiler_optimization::*;
pub use hardware_optimizations::*;
pub use simd::*;
pub use syscall_bypass::*;
pub use zero_copy_io::*;
src/perf/simd.rs
+1 -1
View File
@@ -235,7 +235,7 @@ impl SIMDHash {
/// 批量计算 SHA256 哈希
#[inline(always)]
pub fn hash_batch_sha256(data: &[&[u8]]) -> Vec<[u8; 32]> {
use sha2::{Sha256, Digest};
use sha2::{Digest, Sha256};
data.iter()
.map(|item| {
src/perf/syscall_bypass.rs
+154 -154
View File
@@ -1,11 +1,11 @@
//! 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 std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use anyhow::Result;
use crossbeam_utils::CachePadded;
@@ -55,14 +55,30 @@ pub struct SyscallBatchProcessor {
#[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 },
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 },
MemAlloc {
size: usize,
},
/// 内存释放
MemFree { ptr: usize },
MemFree {
ptr: usize,
},
}
/// 🚀 快速时间提供器 - 绕过系统调用获取时间
@@ -86,24 +102,22 @@ 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
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
)),
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 {
@@ -111,19 +125,19 @@ impl FastTimeProvider {
// 使用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 {
@@ -132,36 +146,34 @@ impl FastTimeProvider {
// 在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
);
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 {
@@ -200,16 +212,16 @@ 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")]
@@ -218,7 +230,7 @@ impl IOOptimizer {
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>() {
@@ -227,60 +239,62 @@ impl IOOptimizer {
}
}
}
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);
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)]
@@ -298,16 +312,12 @@ impl IOOptimizer {
.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 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(),
@@ -317,37 +327,42 @@ impl IOOptimizer {
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,
};
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)),
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),
),
}
}
}
@@ -357,47 +372,46 @@ 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"))?;
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 } => {
@@ -414,28 +428,28 @@ impl SyscallBatchProcessor {
}
}
}
// 批量执行写入
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系统调用进行批量写入
@@ -445,7 +459,7 @@ impl SyscallBatchProcessor {
}
Ok(())
}
/// 批量读取操作
async fn batch_read_operations(&self, requests: Vec<(i32, usize)>) -> Result<()> {
// 使用readv系统调用进行批量读取
@@ -454,7 +468,7 @@ impl SyscallBatchProcessor {
}
Ok(())
}
/// 批量网络操作
async fn batch_network_operations(&self, requests: Vec<(i32, Vec<u8>)>) -> Result<()> {
// 使用sendmsg/recvmsg进行批量网络操作
@@ -482,22 +496,16 @@ impl SystemCallBypassManager {
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,
})
Ok(Self { config, batch_processor, fast_time_provider, _io_optimizer: io_optimizer, stats })
}
/// 🚀 快速获取当前时间戳 - 绕过系统调用
#[inline(always)]
pub fn fast_timestamp_nanos(&self) -> u64 {
@@ -505,28 +513,25 @@ impl SystemCallBypassManager {
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
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> {
@@ -534,34 +539,30 @@ impl SystemCallBypassManager {
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;
use std::sync::Mutex;
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,
})
});
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();
@@ -574,18 +575,18 @@ impl SystemCallBypassManager {
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);
@@ -593,11 +594,11 @@ impl SystemCallBypassManager {
}
}
});
tracing::info!(target: "sol_trade_sdk","✅ Batch processing worker started");
Ok(())
}
/// 获取绕过统计
pub fn get_bypass_stats(&self) -> SyscallBypassStatsSnapshot {
SyscallBypassStatsSnapshot {
@@ -608,7 +609,7 @@ impl SystemCallBypassManager {
memory_operations_avoided: self.stats.memory_operations_avoided.load(Ordering::Relaxed),
}
}
/// 🚀 极致优化配置
pub fn extreme_bypass_config() -> SyscallBypassConfig {
SyscallBypassConfig {
@@ -643,9 +644,11 @@ impl SyscallBypassStatsSnapshot {
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;
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);
}
}
@@ -657,7 +660,7 @@ macro_rules! bypass_syscall {
// 使用快速时间而不是系统调用
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
@@ -667,60 +670,57 @@ macro_rules! bypass_syscall {
#[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]
#[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],
};
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();
@@ -731,16 +731,16 @@ mod tests {
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);
}
}
}
src/perf/zero_copy_io.rs
+133 -126
View File
@@ -1,5 +1,5 @@
//! 🚀 零拷贝内存映射IO - 完全消除数据拷贝开销
//!
//!
//! 实现极致的零拷贝策略，包括：
//! - 内存映射文件IO
//! - 共享内存环形缓冲区
@@ -10,11 +10,11 @@
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
// use std::mem::{size_of, MaybeUninit};
use anyhow::{Context, Result};
use crossbeam_utils::CachePadded;
use memmap2::{MmapMut, MmapOptions};
use std::ptr::NonNull;
use std::slice;
use memmap2::{MmapMut, MmapOptions};
use anyhow::{Result, Context};
use crossbeam_utils::CachePadded;
/// 🚀 零拷贝内存管理器
pub struct ZeroCopyMemoryManager {
@@ -50,17 +50,17 @@ impl SharedMemoryPool {
// 确保块大小是64字节对齐(缓存行对齐)
let aligned_block_size = (block_size + 63) & !63;
let total_blocks = total_size / aligned_block_size;
// 创建内存映射文件
let memory_region = MmapOptions::new()
.len(total_blocks * aligned_block_size)
.map_anon()
.context("Failed to create memory mapped region")?;
// 初始化空闲块位图 (每个u64可以管理64个块)
let bitmap_size = (total_blocks + 63) / 64;
let mut free_blocks = Vec::with_capacity(bitmap_size);
// 将所有块标记为空闲(全1)
for i in 0..bitmap_size {
let bits = if i == bitmap_size - 1 && total_blocks % 64 != 0 {
@@ -72,10 +72,10 @@ impl SharedMemoryPool {
};
free_blocks.push(AtomicU64::new(bits));
}
tracing::info!(target: "sol_trade_sdk","🚀 Created shared memory pool {} with {} blocks of {} bytes each",
pool_id, total_blocks, aligned_block_size);
Ok(Self {
memory_region,
free_blocks,
@@ -85,31 +85,31 @@ impl SharedMemoryPool {
pool_id,
})
}
/// 🚀 零拷贝分配内存块
#[inline(always)]
pub fn allocate_block(&self) -> Option<ZeroCopyBlock> {
// 快速路径：尝试从预期位置分配
let start_index = self.allocator_head.load(Ordering::Relaxed) / 64;
// 遍历所有位图寻找空闲块
for attempt in 0..self.free_blocks.len() {
let bitmap_index = (start_index + attempt) % self.free_blocks.len();
let bitmap = &self.free_blocks[bitmap_index];
let mut current = bitmap.load(Ordering::Acquire);
while current != 0 {
// 找到最低位的1(最小的空闲块)
let bit_pos = current.trailing_zeros() as usize;
let mask = 1u64 << bit_pos;
// 尝试原子地清除这一位(标记为已分配)
match bitmap.compare_exchange_weak(
current,
current,
current & !mask,
Ordering::AcqRel,
Ordering::Relaxed
Ordering::Relaxed,
) {
Ok(_) => {
// 成功分配
@@ -119,20 +119,17 @@ impl SharedMemoryPool {
bitmap.fetch_or(mask, Ordering::Relaxed);
break;
}
let offset = block_index * self.block_size;
let ptr = unsafe {
NonNull::new_unchecked(
self.memory_region.as_ptr().add(offset) as *mut u8
)
};
// 更新分配器头指针
self.allocator_head.store(
(block_index + 1) * 64,
Ordering::Relaxed
);
self.allocator_head.store((block_index + 1) * 64, Ordering::Relaxed);
return Some(ZeroCopyBlock {
ptr,
size: self.block_size,
@@ -147,10 +144,10 @@ impl SharedMemoryPool {
}
}
}
None // 没有可用块
}
/// 🚀 零拷贝释放内存块
#[inline(always)]
pub fn deallocate_block(&self, block: ZeroCopyBlock) {
@@ -158,20 +155,21 @@ impl SharedMemoryPool {
tracing::error!(target: "sol_trade_sdk", "Attempting to deallocate block from wrong pool");
return;
}
let bitmap_index = block.block_index / 64;
let bit_pos = block.block_index % 64;
let mask = 1u64 << bit_pos;
if bitmap_index < self.free_blocks.len() {
// 原子地设置位为1(标记为空闲)
self.free_blocks[bitmap_index].fetch_or(mask, Ordering::Release);
}
}
/// 获取可用块数量
pub fn available_blocks(&self) -> usize {
self.free_blocks.iter()
self.free_blocks
.iter()
.map(|bitmap| bitmap.load(Ordering::Relaxed).count_ones() as usize)
.sum()
}
@@ -195,49 +193,49 @@ impl ZeroCopyBlock {
pub fn as_ptr(&self) -> *mut u8 {
self.ptr.as_ptr()
}
/// 获取只读切片
#[inline(always)]
pub unsafe fn as_slice(&self) -> &[u8] {
slice::from_raw_parts(self.ptr.as_ptr(), self.size)
}
/// 获取可变切片
#[inline(always)]
pub unsafe fn as_mut_slice(&mut self) -> &mut [u8] {
slice::from_raw_parts_mut(self.ptr.as_ptr(), self.size)
}
/// 获取块大小
#[inline(always)]
pub fn size(&self) -> usize {
self.size
}
/// 零拷贝写入数据
#[inline(always)]
pub unsafe fn write_bytes(&mut self, data: &[u8]) -> Result<()> {
if data.len() > self.size {
return Err(anyhow::anyhow!("Data too large for block"));
}
// 使用硬件优化的内存拷贝
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
self.ptr.as_ptr(),
data.as_ptr(),
data.len()
data.len(),
);
Ok(())
}
/// 零拷贝读取数据
#[inline(always)]
pub unsafe fn read_bytes(&self, len: usize) -> Result<&[u8]> {
if len > self.size {
return Err(anyhow::anyhow!("Read length exceeds block size"));
}
Ok(slice::from_raw_parts(self.ptr.as_ptr(), len))
}
}
@@ -266,9 +264,9 @@ impl MemoryMappedBuffer {
.len(size)
.map_anon()
.context("Failed to create memory mapped buffer")?;
tracing::info!(target: "sol_trade_sdk","🚀 Created memory mapped buffer {} with size {} bytes", buffer_id, size);
Ok(Self {
mmap,
read_pos: CachePadded::new(AtomicUsize::new(0)),
@@ -277,128 +275,136 @@ impl MemoryMappedBuffer {
_buffer_id: buffer_id,
})
}
/// 🚀 零拷贝写入数据
#[inline(always)]
pub fn write_data(&self, data: &[u8]) -> Result<usize> {
let data_len = data.len();
let current_write = self.write_pos.load(Ordering::Relaxed);
let current_read = self.read_pos.load(Ordering::Acquire);
// 计算可用空间
let available_space = if current_write >= current_read {
self.size - (current_write - current_read) - 1
} else {
current_read - current_write - 1
};
if data_len > available_space {
return Err(anyhow::anyhow!("Insufficient buffer space"));
}
// 零拷贝写入
unsafe {
let write_ptr = self.mmap.as_ptr().add(current_write) as *mut u8;
if current_write + data_len <= self.size {
// 数据不跨越缓冲区边界
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
write_ptr, data.as_ptr(), data_len
write_ptr,
data.as_ptr(),
data_len,
);
} else {
// 数据跨越缓冲区边界，分两段写入
let first_part = self.size - current_write;
let second_part = data_len - first_part;
// 写入第一部分
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
write_ptr, data.as_ptr(), first_part
write_ptr,
data.as_ptr(),
first_part,
);
// 写入第二部分(从缓冲区开头)
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
self.mmap.as_ptr() as *mut u8,
data.as_ptr().add(first_part),
second_part
self.mmap.as_ptr() as *mut u8,
data.as_ptr().add(first_part),
second_part,
);
}
}
// 更新写指针
let new_write_pos = (current_write + data_len) % self.size;
self.write_pos.store(new_write_pos, Ordering::Release);
Ok(data_len)
}
/// 🚀 零拷贝读取数据
#[inline(always)]
pub fn read_data(&self, buffer: &mut [u8]) -> Result<usize> {
let buffer_len = buffer.len();
let current_read = self.read_pos.load(Ordering::Relaxed);
let current_write = self.write_pos.load(Ordering::Acquire);
// 计算可读数据量
let available_data = if current_write >= current_read {
current_write - current_read
} else {
self.size - (current_read - current_write)
};
if available_data == 0 {
return Ok(0); // 无数据可读
}
let read_len = buffer_len.min(available_data);
// 零拷贝读取
unsafe {
let read_ptr = self.mmap.as_ptr().add(current_read);
if current_read + read_len <= self.size {
// 数据不跨越缓冲区边界
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
buffer.as_mut_ptr(), read_ptr, read_len
buffer.as_mut_ptr(),
read_ptr,
read_len,
);
} else {
// 数据跨越缓冲区边界，分两段读取
let first_part = self.size - current_read;
let second_part = read_len - first_part;
// 读取第一部分
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
buffer.as_mut_ptr(), read_ptr, first_part
buffer.as_mut_ptr(),
read_ptr,
first_part,
);
// 读取第二部分(从缓冲区开头)
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
buffer.as_mut_ptr().add(first_part),
self.mmap.as_ptr(),
second_part
self.mmap.as_ptr(),
second_part,
);
}
}
// 更新读指针
let new_read_pos = (current_read + read_len) % self.size;
self.read_pos.store(new_read_pos, Ordering::Release);
Ok(read_len)
}
/// 获取可读数据量
#[inline(always)]
pub fn available_data(&self) -> usize {
let current_read = self.read_pos.load(Ordering::Relaxed);
let current_write = self.write_pos.load(Ordering::Relaxed);
if current_write >= current_read {
current_write - current_read
} else {
self.size - (current_read - current_write)
}
}
/// 获取可用空间
#[inline(always)]
pub fn available_space(&self) -> usize {
@@ -420,38 +426,39 @@ impl DirectMemoryAccessManager {
/// 创建DMA管理器
pub fn new(num_channels: usize) -> Result<Self> {
let mut dma_channels = Vec::with_capacity(num_channels);
for i in 0..num_channels {
dma_channels.push(Arc::new(DMAChannel::new(i)?));
}
tracing::info!(target: "sol_trade_sdk","🚀 Created DMA manager with {} channels", num_channels);
Ok(Self {
dma_channels,
channel_allocator: AtomicUsize::new(0),
dma_stats: Arc::new(DMAStats::new()),
})
}
/// 🚀 执行零拷贝DMA传输
#[inline(always)]
pub async fn dma_transfer(&self, src: &[u8], dst: &mut [u8]) -> Result<usize> {
if src.len() != dst.len() {
return Err(anyhow::anyhow!("Source and destination sizes don't match"));
}
// 选择DMA通道(轮询分配)
let channel_index = self.channel_allocator.fetch_add(1, Ordering::Relaxed) % self.dma_channels.len();
let channel_index =
self.channel_allocator.fetch_add(1, Ordering::Relaxed) % self.dma_channels.len();
let channel = &self.dma_channels[channel_index];
// 执行DMA传输
let transferred = channel.transfer(src, dst).await?;
// 更新统计
self.dma_stats.bytes_transferred.fetch_add(transferred as u64, Ordering::Relaxed);
self.dma_stats.transfers_completed.fetch_add(1, Ordering::Relaxed);
Ok(transferred)
}
}
@@ -475,21 +482,21 @@ impl DMAChannel {
_status: AtomicU64::new(0),
})
}
/// 🚀 执行零拷贝传输
#[inline(always)]
pub async fn transfer(&self, src: &[u8], dst: &mut [u8]) -> Result<usize> {
let transfer_size = src.len();
// 使用硬件优化的SIMD内存拷贝
unsafe {
super::hardware_optimizations::SIMDMemoryOps::memcpy_simd_optimized(
dst.as_mut_ptr(),
src.as_ptr(),
transfer_size
transfer_size,
);
}
Ok(transfer_size)
}
}
@@ -541,14 +548,14 @@ impl ZeroCopyStats {
mmap_buffer_usage: AtomicU64::new(0),
}
}
/// 打印统计信息
pub fn print_stats(&self) {
let allocated = self.blocks_allocated.load(Ordering::Relaxed);
let freed = self.blocks_freed.load(Ordering::Relaxed);
let bytes = self.bytes_transferred.load(Ordering::Relaxed);
let mmap_usage = self.mmap_buffer_usage.load(Ordering::Relaxed);
tracing::info!(target: "sol_trade_sdk","🚀 Zero-Copy Stats:");
tracing::info!(target: "sol_trade_sdk"," 📦 Blocks: Allocated={}, Freed={}, Active={}",
allocated, freed, allocated.saturating_sub(freed));
@@ -564,36 +571,36 @@ impl ZeroCopyMemoryManager {
pub fn new() -> Result<Self> {
let mut shared_pools = Vec::new();
let mut mmap_buffers = Vec::new();
// 创建不同大小的内存池
// 小块池: 64KB blocks, 1GB total
shared_pools.push(Arc::new(SharedMemoryPool::new(0, 1024 * 1024 * 1024, 64 * 1024)?));
// 中块池: 1MB blocks, 4GB total
// 中块池: 1MB blocks, 4GB total
shared_pools.push(Arc::new(SharedMemoryPool::new(1, 4 * 1024 * 1024 * 1024, 1024 * 1024)?));
// 大块池: 16MB blocks, 8GB total
shared_pools.push(Arc::new(SharedMemoryPool::new(2, 8 * 1024 * 1024 * 1024, 16 * 1024 * 1024)?));
shared_pools.push(Arc::new(SharedMemoryPool::new(
2,
8 * 1024 * 1024 * 1024,
16 * 1024 * 1024,
)?));
// 创建内存映射缓冲区
for i in 0..8 {
mmap_buffers.push(Arc::new(MemoryMappedBuffer::new(i, 256 * 1024 * 1024)?)); // 256MB each
mmap_buffers.push(Arc::new(MemoryMappedBuffer::new(i, 256 * 1024 * 1024)?));
// 256MB each
}
let dma_manager = Arc::new(DirectMemoryAccessManager::new(16)?); // 16 DMA channels
let stats = Arc::new(ZeroCopyStats::new());
tracing::info!(target: "sol_trade_sdk","🚀 Zero-Copy Memory Manager initialized");
tracing::info!(target: "sol_trade_sdk"," 📦 Memory Pools: {}", shared_pools.len());
tracing::info!(target: "sol_trade_sdk"," 💾 Mapped Buffers: {}", mmap_buffers.len());
tracing::info!(target: "sol_trade_sdk"," 🔄 DMA Channels: 16");
Ok(Self {
shared_pools,
mmap_buffers,
dma_manager,
stats,
})
Ok(Self { shared_pools, mmap_buffers, dma_manager, stats })
}
/// 🚀 分配零拷贝内存块
#[inline(always)]
pub fn allocate(&self, size: usize) -> Option<ZeroCopyBlock> {
@@ -605,7 +612,7 @@ impl ZeroCopyMemoryManager {
} else {
&self.shared_pools[2] // 大块池
};
if let Some(block) = pool.allocate_block() {
self.stats.blocks_allocated.fetch_add(1, Ordering::Relaxed);
Some(block)
@@ -613,7 +620,7 @@ impl ZeroCopyMemoryManager {
None
}
}
/// 🚀 释放零拷贝内存块
#[inline(always)]
pub fn deallocate(&self, block: ZeroCopyBlock) {
@@ -623,19 +630,19 @@ impl ZeroCopyMemoryManager {
self.stats.blocks_freed.fetch_add(1, Ordering::Relaxed);
}
}
/// 获取内存映射缓冲区
#[inline(always)]
pub fn get_mmap_buffer(&self, buffer_id: usize) -> Option<Arc<MemoryMappedBuffer>> {
self.mmap_buffers.get(buffer_id).cloned()
}
/// 获取DMA管理器
#[inline(always)]
pub fn get_dma_manager(&self) -> Arc<DirectMemoryAccessManager> {
self.dma_manager.clone()
}
/// 获取统计信息
pub fn get_stats(&self) -> Arc<ZeroCopyStats> {
self.stats.clone()
@@ -645,73 +652,73 @@ impl ZeroCopyMemoryManager {
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_shared_memory_pool() -> Result<()> {
let pool = SharedMemoryPool::new(0, 1024 * 1024, 4096)?;
// 测试分配
let block1 = pool.allocate_block().expect("Should allocate block");
assert_eq!(block1.size(), 4096);
let block2 = pool.allocate_block().expect("Should allocate another block");
assert_eq!(block2.size(), 4096);
// 测试释放
pool.deallocate_block(block1);
pool.deallocate_block(block2);
Ok(())
}
#[tokio::test]
async fn test_memory_mapped_buffer() -> Result<()> {
let buffer = MemoryMappedBuffer::new(0, 1024 * 1024)?;
let test_data = b"Hello, Zero-Copy World!";
// 测试写入
let written = buffer.write_data(test_data)?;
assert_eq!(written, test_data.len());
// 测试读取
let mut read_buffer = vec![0u8; test_data.len()];
let read = buffer.read_data(&mut read_buffer)?;
assert_eq!(read, test_data.len());
assert_eq!(&read_buffer, test_data);
Ok(())
}
#[tokio::test]
async fn test_dma_transfer() -> Result<()> {
let dma_manager = DirectMemoryAccessManager::new(4)?;
let src = vec![1u8, 2, 3, 4, 5, 6, 7, 8];
let mut dst = vec![0u8; 8];
let transferred = dma_manager.dma_transfer(&src, &mut dst).await?;
assert_eq!(transferred, 8);
assert_eq!(src, dst);
Ok(())
}
#[tokio::test]
async fn test_zero_copy_manager() -> Result<()> {
let manager = ZeroCopyMemoryManager::new()?;
// 测试小块分配
let small_block = manager.allocate(1024).expect("Should allocate small block");
assert_eq!(small_block.size(), 65536); // 小块池的块大小
// 测试大块分配
let large_block = manager.allocate(5 * 1024 * 1024).expect("Should allocate large block");
assert_eq!(large_block.size(), 16 * 1024 * 1024); // 大块池的块大小
manager.deallocate(small_block);
manager.deallocate(large_block);
Ok(())
}
}
}