Rust 异步 IO：从 epoll 到 io_uring(最新推荐)

应用代码(async fn)
    ↓ 编译
状态机 Future
    ↓ poll()
Tokio 运行时
    ↓ 注册 fd
epoll/io_uring
    ↓
Linux 内核

use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Semaphore;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use std::sync::Arc;
use std::time::Duration;
struct ServerConfig {
    max_connections: usize,
    read_buffer_size: usize,
    write_timeout: Duration,
}
async fn handle_connection(
    mut stream: TcpStream,
    config: Arc<ServerConfig>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut buffer = vec![0u8; config.read_buffer_size];
    loop {
        let n = tokio::time::timeout(
            Duration::from_secs(30),
            stream.read(&mut buffer)
        ).await??;
        if n == 0 {
            break;
        }
        tokio::time::timeout(
            config.write_timeout,
            stream.write_all(&buffer[..n])
        ).await??;
    }
    Ok(())
}
async fn run_server(config: ServerConfig) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let listener = TcpListener::bind("0.0.0.0:8080").await?;
    let config = Arc::new(config);
    let semaphore = Arc::new(Semaphore::new(config.max_connections));
    println!("服务启动，最大并发连接: {}", config.max_connections);
    loop {
        let (stream, addr) = listener.accept().await?;
        let permit = semaphore.clone().acquire_owned().await?;
        let config = config.clone();
        tokio::spawn(async move {
            let _permit = permit;
            if let Err(e) = handle_connection(stream, config).await {
                eprintln!("连接 {} 处理异常: {}", addr, e);
            }
        });
    }
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let config = ServerConfig {
        max_connections: 10000,
        read_buffer_size: 8192,
        write_timeout: Duration::from_secs(10),
    };
    run_server(config).await
}

use tokio_uring::net::TcpListener;
use tokio_uring::buf::IoBufMut;
async fn handle_connection_uring(
    stream: tokio_uring::net::TcpStream,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let buffer = vec![0u8; 8192];
    loop {
        let (n, buffer) = stream.read(buffer).await?;
        if n == 0 {
            break;
        }
        let (n, buffer) = stream.write_all(buffer[..n]).await?;
        drop(buffer);
    }
    Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let listener = TcpListener::bind("0.0.0.0:8081").await?;
    println!("io_uring 服务启动");
    loop {
        let (stream, addr) = listener.accept().await?;
        tokio_uring::spawn(async move {
            if let Err(e) = handle_connection_uring(stream).await {
                eprintln!("连接 {} 处理异常: {}", addr, e);
            }
        });
    }
}