Rust 系统编程：构建高性能、内存安全的基础设施

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use std::future::Future;
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

// 简化的任务结构
struct Task {
    id: usize,
    future: Pin<Box<dyn Future<Output = ()>>>,
}

// 自定义执行器
struct Executor {
    tasks: Vec<Task>,
    task_queue: Arc<crossbeam::queue::SegQueue<Task>>,
}

impl Executor {
    fn new() -> Self {
        Executor {
            tasks: Vec::new(),
            task_queue: Arc::new(crossbeam::queue::SegQueue::new()),
        }
    }

    fn spawn<F>(&mut self, future: F)
    where
        F: Future<Output = ()> + 'static,
    {
        let task = Task {
            id: self.tasks.len(),
            future: Box::pin(future),
        };
        self.tasks.push(task);
    }

    fn run(&mut self) {
        // 创建虚拟 waker
        let waker = self.create_dummy_waker();
        let mut cx = Context::from_waker(&waker);

        // 轮询所有任务
        for task in &mut self.tasks {
            match task.future.as_mut().poll(&mut cx) {
                Poll::Ready(()) => {
                    println!("Task {} completed", task.id);
                }
                Poll::Pending => {
                    println!("Task {} is pending", task.id);
                }
            }
        }
    }

    fn create_dummy_waker(&self) -> Waker {
        unsafe fn clone(raw: *const ()) -> RawWaker {
            RawWaker::new(raw, &VTABLE)
        }

        unsafe fn wake(_raw: *const ()) {
            // 唤醒逻辑
        }

        unsafe fn wake_by_ref(_raw: *const ()) {
            // 唤醒逻辑
        }

        unsafe fn drop(_raw: *const ()) {
            // 清理逻辑
        }

        const VTABLE: RawWakerVTable = RawWakerVTable::new(
            clone,
            wake,
            wake_by_ref,
            drop,
        );

        let raw = std::ptr::null();
        let raw_waker = RawWaker::new(raw, &VTABLE);
        unsafe { Waker::from_raw(raw_waker) }
    }
}

// 使用示例
async fn hello_world() {
    println!("Hello from async!");
}

fn main() {
    let mut executor = Executor::new();
    executor.spawn(hello_world());
    executor.run();
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166

use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use bytes::{Buf, BufMut, BytesMut};
use std::error::Error;

// 零拷贝缓冲区
struct ZeroCopyBuffer {
    buffer: BytesMut,
    read_pos: usize,
}

impl ZeroCopyBuffer {
    fn new(capacity: usize) -> Self {
        ZeroCopyBuffer {
            buffer: BytesMut::with_capacity(capacity),
            read_pos: 0,
        }
    }

    // 读取数据到缓冲区（零拷贝）
    async fn read_from_stream(
        &mut self,
        stream: &mut TcpStream
    ) -> Result<usize, Box<dyn Error>> {
        // 确保有足够空间
        self.buffer.reserve(8192);

        let mut temp = [0u8; 8192];
        let n = stream.read(&mut temp).await?;

        // 追加到缓冲区（零拷贝）
        self.buffer.put_slice(&temp[..n]);

        Ok(n)
    }

    // 获取读视图（零拷贝）
    fn get_read_view(&self) -> &[u8] {
        &self.buffer[self.read_pos..]
    }

    // 消费已读取的数据
    fn consume(&mut self, n: usize) {
        self.read_pos += n;

        // 清理已消费的数据
        if self.read_pos > self.buffer.len() / 2 {
            self.buffer.advance(self.read_pos);
            self.read_pos = 0;
        }
    }
}

// 高性能 HTTP 服务器
struct HttpServer {
    listener: TcpListener,
    buffer_pool: Arc<Mutex<Vec<BytesMut>>>,
}

impl HttpServer {
    async fn new(addr: &str) -> Result<Self, Box<dyn Error>> {
        Ok(HttpServer {
            listener: TcpListener::bind(addr).await?,
            buffer_pool: Arc::new(Mutex::new(Vec::new())),
        })
    }

    async fn run(&self) -> Result<(), Box<dyn Error>> {
        loop {
            let (mut stream, addr) = self.listener.accept().await?;

            println!("New connection from {}", addr);

            let buffer_pool = self.buffer_pool.clone();
            tokio::spawn(async move {
                if let Err(e) = Self::handle_connection(&mut stream, buffer_pool).await {
                    eprintln!("Error handling connection: {}", e);
                }
            });
        }
    }

    async fn handle_connection(
        stream: &mut TcpStream,
        buffer_pool: Arc<Mutex<Vec<BytesMut>>>,
    ) -> Result<(), Box<dyn Error>> {
        // 从池中获取缓冲区
        let mut buffer = {
            let mut pool = buffer_pool.lock().await;
            pool.pop().unwrap_or_else(|| BytesMut::with_capacity(8192))
        };

        let mut read_buffer = ZeroCopyBuffer::new(8192);

        loop {
            // 读取请求
            let n = read_buffer.read_from_stream(stream).await?;
            if n == 0 {
                break;
            }

            // 解析 HTTP 请求
            let request = Self::parse_request(read_buffer.get_read_view())?;
            read_buffer.consume(request.raw_len);

            // 生成响应
            buffer.clear();
            Self::generate_response(&request, &mut buffer);

            // 发送响应（零拷贝）
            stream.write_all(&buffer[..]).await?;
        }

        // 归还缓冲区到池
        {
            let mut pool = buffer_pool.lock().await;
            pool.push(buffer);
        }

        Ok(())
    }

    fn parse_request(data: &[u8]) -> Result<HttpRequest, Box<dyn Error>> {
        // 简化的 HTTP 请求解析
        let request_str = std::str::from_utf8(data)?;

        if let Some(line) = request_str.lines().next() {
            let parts: Vec<&str> = line.split_whitespace().collect();
            if parts.len() >= 2 {
                return Ok(HttpRequest {
                    method: parts[0].to_string(),
                    path: parts[1].to_string(),
                    version: if parts.len() > 2 {
                        parts[2].to_string()
                    } else {
                        "HTTP/1.1".to_string()
                    },
                    raw_len: line.len() + 2, // +2 for CRLF
                });
            }
        }

        Err("Invalid HTTP request".into())
    }

    fn generate_response(request: &HttpRequest, buffer: &mut BytesMut) {
        let response = format!(
            "HTTP/1.1 200 OK\r\n\
             Content-Type: text/html\r\n\
             Content-Length: {}\r\n\
             \r\n\
             <html><body><h1>Hello, {}!</h1></body></html>",
            request.path.len() + 28, // 粗略计算内容长度
            request.path
        );

        buffer.put_slice(response.as_bytes());
    }
}

struct HttpRequest {
    method: String,
    path: String,
    version: String,
    raw_len: usize,
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

use std::io;
use std::os::unix::io::{AsRawFd, RawFd};
use libc::{epoll_create1, epoll_ctl, epoll_wait, epoll_event, EPOLLIN, EPOLL_CTL_ADD, EPOLL_CLOEXEC};

// Epoll 封装
struct Epoll {
    fd: RawFd,
}

impl Epoll {
    fn new() -> io::Result<Self> {
        let fd = unsafe { epoll_create1(EPOLL_CLOEXEC) };

        if fd == -1 {
            return Err(io::Error::last_os_error());
        }

        Ok(Epoll { fd })
    }

    fn add(&self, target_fd: RawFd, data: u64) -> io::Result<()> {
        let mut event = epoll_event {
            events: EPOLLIN as u32,
            u64: data,
        };

        let result = unsafe {
            epoll_ctl(
                self.fd,
                EPOLL_CTL_ADD,
                target_fd,
                &mut event,
            )
        };

        if result == -1 {
            return Err(io::Error::last_os_error());
        }

        Ok(())
    }

    fn wait(&self, timeout_ms: i32) -> io::Result<Vec<u64>> {
        const MAX_EVENTS: usize = 32;
        let mut events: [epoll_event; MAX_EVENTS] = unsafe { std::mem::zeroed() };

        let count = unsafe {
            epoll_wait(
                self.fd,
                events.as_mut_ptr(),
                MAX_EVENTS as i32,
                timeout_ms,
            )
        };

        if count == -1 {
            return Err(io::Error::last_os_error());
        }

        let mut ready = Vec::with_capacity(count as usize);
        for i in 0..count as usize {
            ready.push(unsafe { events[i].u64 });
        }

        Ok(ready)
    }
}

impl Drop for Epoll {
    fn drop(&mut self) {
        unsafe {
            libc::close(self.fd);
        }
    }
}

// 反应器模式
struct Reactor {
    epoll: Epoll,
    handles: Vec<tokio::sync::oneshot::Sender<()>>,
}

impl Reactor {
    fn new() -> io::Result<Self> {
        Ok(Reactor {
            epoll: Epoll::new()?,
            handles: Vec::new(),
        })
    }

    fn register(&mut self, fd: RawFd) -> tokio::sync::oneshot::Receiver<()> {
        let (tx, rx) = tokio::sync::oneshot::channel();
        let data = self.handles.len() as u64;

        self.epoll.add(fd, data).unwrap();
        self.handles.push(tx);

        rx
    }

    async fn wait_event(&mut self) -> io::Result<()> {
        loop {
            let ready = self.epoll.wait(100)?;

            for data in ready {
                if let Some(tx) = self.handles.get(data as usize) {
                    let _ = tx.send(());
                }
            }

            if !ready.is_empty() {
                return Ok(());
            }
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};

// 内存池分配器
struct PoolAllocator {
    // 简化的内存池实现
    pool: Vec<Vec<u8>>,
    allocations: AtomicUsize,
}

unsafe impl GlobalAlloc for PoolAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        self.allocations.fetch_add(1, Ordering::Relaxed);

        // 根据大小选择合适的内存块
        let size = layout.size();
        let align = layout.align();

        // 简化：直接使用系统分配
        System.alloc(layout)
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        self.allocations.fetch_sub(1, Ordering::Relaxed);
        System.dealloc(ptr, layout)
    }
}

//  arenas 分配器
use bumpalo::Bump;

struct ArenaAllocator {
    arena: Bump,
}

impl ArenaAllocator {
    fn new() -> Self {
        ArenaAllocator {
            arena: Bump::new(),
        }
    }

    fn alloc<T>(&self, value: T) -> &T {
        self.arena.alloc(value)
    }

    fn alloc_str(&self, s: &str) -> &str {
        self.arena.alloc_str(s)
    }

    fn alloc_slice<T>(&self, slice: &[T]) -> &[T]
    where
        T: Copy,
    {
        self.arena.alloc_slice_copy(slice)
    }

    fn reset(&self) {
        self.arena.reset();
    }
}

// 使用示例
fn process_with_arena(data: &[u8]) -> String {
    let arena = ArenaAllocator::new();

    // 在 arena 中分配大量临时对象
    let parts: Vec<&str> = data
        .split(|&b| b == b' ')
        .map(|bytes| {
            std::str::from_utf8(bytes).unwrap_or("")
        })
        .map(|s| arena.alloc_str(s))
        .collect();

    // 处理数据
    let result = parts.join(" ");

    // arena 会在作用域结束时自动清理
    result
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

use std::borrow::Cow;

// 零拷贝字符串处理
fn process_string_cow(input: &str) -> Cow<str> {
    if input.contains("need_processing") {
        // 需要处理时返回拥有的字符串
        Cow::Owned(input.to_uppercase())
    } else {
        // 不需要处理时返回借用（零拷贝）
        Cow::Borrowed(input)
    }
}

// bytes 类型的零拷贝
use bytes::{Bytes, BytesMut};

struct Message {
    // Bytes 可以引用共享内存而不复制
    data: Bytes,
}

impl Message {
    // 从字节切片创建（可能涉及复制）
    fn from_slice(data: &[u8]) -> Self {
        Message {
            data: Bytes::copy_from_slice(data),
        }
    }

    // 从 BytesMut 创建（零拷贝）
    fn from_bytes_mut(data: BytesMut) -> Self {
        Message {
            data: data.freeze(),
        }
    }

    // 切片（零拷贝）
    fn slice(&self, start: usize, end: usize) -> Message {
        Message {
            data: self.data.slice(start, end),
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

use std::sync::atomic::{AtomicUsize, Ordering};

// 无锁计数器
struct LockFreeCounter {
    count: AtomicUsize,
}

impl LockFreeCounter {
    fn new() -> Self {
        LockFreeCounter {
            count: AtomicUsize::new(0),
        }
    }

    fn increment(&self) -> usize {
        self.count.fetch_add(1, Ordering::Relaxed)
    }

    fn get(&self) -> usize {
        self.count.load(Ordering::Relaxed)
    }
}

// 无锁队列（简化版）
use crossbeam::queue::SegQueue;
use std::sync::Arc;

struct LockFreeQueue<T> {
    queue: Arc<SegQueue<T>>,
}

impl<T> LockFreeQueue<T> {
    fn new() -> Self {
        LockFreeQueue {
            queue: Arc::new(SegQueue::new()),
        }
    }

    fn push(&self, item: T) {
        self.queue.push(item);
    }

    fn try_pop(&self) -> Option<T> {
        self.queue.pop()
    }

    fn clone(&self) -> Self {
        LockFreeQueue {
            queue: Arc::clone(&self.queue),
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

use tokio::sync::mpsc;

// 消息类型
enum Message {
    Task(String),
    Stop,
}

// Actor 结构
struct Actor {
    receiver: mpsc::Receiver<Message>,
    state: String,
}

impl Actor {
    fn new(receiver: mpsc::Receiver<Message>) -> Self {
        Actor {
            receiver,
            state: String::new(),
        }
    }

    async fn run(mut self) {
        while let Some(msg) = self.receiver.recv().await {
            match msg {
                Message::Task(task) => {
                    self.handle_task(task).await;
                }
                Message::Stop => {
                    println!("Actor stopping...");
                    break;
                }
            }
        }
    }

    async fn handle_task(&mut self, task: String) {
        println!("Processing task: {}", task);
        self.state = task;
    }
}

// Actor 管理器
struct ActorManager {
    senders: Vec<mpsc::Sender<Message>>,
}

impl ActorManager {
    fn new(num_actors: usize) -> Self {
        let mut senders = Vec::new();

        for _ in 0..num_actors {
            let (tx, rx) = mpsc::channel(100);
            let actor = Actor::new(rx);

            tokio::spawn(actor.run());

            senders.push(tx);
        }

        ActorManager { senders }
    }

    fn send_task(&self, task: String) -> Result<(), mpsc::error::SendError<Message>> {
        // 简单的轮询策略
        let idx = task.len() % self.senders.len();
        self.senders[idx].send(Message::Task(task))
    }

    async fn stop_all(&self) {
        for sender in &self.senders {
            let _ = sender.send(Message::Stop).await;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

use std::arch::x86_64::*;

// SIMD 向量加法
#[cfg(target_arch = "x86_64")]
fn vector_add_simd(a: &[f32], b: &[f32]) -> Vec<f32> {
    assert_eq!(a.len(), b.len());

    let mut result = vec![0.0f32; a.len()];
    let chunks = a.len() / 8;

    unsafe {
        for i in 0..chunks {
            let a_vec = _mm256_loadu_ps(a.as_ptr().add(i * 8));
            let b_vec = _mm256_loadu_ps(b.as_ptr().add(i * 8));
            let result_vec = _mm256_add_ps(a_vec, b_vec);
            _mm256_storeu_ps(result.as_mut_ptr().add(i * 8), result_vec);
        }
    }

    // 处理剩余元素
    for i in (chunks * 8)..a.len() {
        result[i] = a[i] + b[i];
    }

    result
}

// 字符串搜索优化
#[cfg(target_arch = "x86_64")]
fn contains_pattern_simd(text: &[u8], pattern: u8) -> bool {
    const LANE_SIZE: usize = 32;

    unsafe {
        let pattern_vec = _mm256_set1_epi8(pattern as i8);

        let mut i = 0;
        while i + LANE_SIZE <= text.len() {
            let text_vec = _mm256_loadu_si256(text.as_ptr().add(i) as *const __m256i);
            let cmp = _mm256_cmpeq_epi8(text_vec, pattern_vec);
            let mask = _mm256_movemask_epi8(cmp);

            if mask != 0 {
                return true;
            }

            i += LANE_SIZE;
        }

        // 检查剩余字节
        text[i..].contains(&pattern)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

// 结构体布局优化
#[repr(C)]
struct CacheFriendlyStruct {
    // 频繁访问的字段放在一起
    hot_field1: u64,
    hot_field2: u64,
    hot_field3: u64,

    // 不常访问的字段
    #[allow(dead_code)]
    cold_field1: u64,
    #[allow(dead_code)]
    cold_field2: u64,
}

// 数组结构 vs 结构数组
struct AoS {
    x: Vec<f32>,
    y: Vec<f32>,
    z: Vec<f32>,
}

struct SoA {
    x: Vec<f32>,
    y: Vec<f32>,
    z: Vec<f32>,
}

// SoA 更适合向量化处理
fn process_soa(data: &SoA) -> Vec<f32> {
    data.x.iter()
        .zip(&data.y)
        .zip(&data.z)
        .map(|((&x, &y), &z)| (x * x + y * y + z * z).sqrt())
        .collect()
}