Rust线程池库affinitypool的使用：高效管理CPU亲和性与任务调度的并发工具

介绍

affinitypool是一个Rust线程池库，专注于高效管理CPU亲和性和任务调度。它允许你将特定任务绑定到特定的CPU核心上运行，从而优化缓存利用率并减少线程迁移开销，特别适合高性能计算和低延迟应用场景。

主要特性

• 可配置的线程池大小
• 支持设置CPU亲和性
• 任务优先级调度
• 轻量级实现
• 与标准库Future兼容

使用方法

基本使用

首先在Cargo.toml中添加依赖：

[dependencies]
affinitypool = "0.2"

创建线程池

use affinitypool::AffinityPool;

fn main() {
    // 创建4个工作线程的线程池
    let pool = AffinityPool::new(4).expect("Failed to create thread pool");
    
    // 提交任务
    let result = pool.run(|| {
        // 这里执行计算密集型任务
        42
    }).expect("Failed to submit task");
    
    println!("Task result: {}", result);
}

设置CPU亲和性

use affinitypool::AffinityPool;
use core_affinity::CoreId;

fn main() {
    // 获取可用的CPU核心
    let core_ids = core_affinity::get_core_ids().unwrap();
    
    // 创建线程池并设置亲和性
    let pool = AffinityPool::with_affinity(core_ids.len(), |thread_idx| {
        Some(core_ids[thread_idx % core_ids.len()])
    }).expect("Failed to create thread pool with affinity");
    
    // 提交多个任务
    let handles: Vec<_> = (0..10).map(|i| {
        pool.run(move || {
            println!("Task {} running on thread {:?}", i, std::thread::current().id());
            i * i
        }).unwrap()
    }).collect();
    
    // 等待所有任务完成
    for handle in handles {
        println!("Result: {}", handle.join().unwrap());
    }
}

使用异步任务

use affinitypool::AffinityPool;
use futures::future::join_all;

#[tokio::main]
async fn main() {
    let pool = AffinityPool::new(4).unwrap();
    
    let futures: Vec<_> = (0..10).map(|i| {
        pool.spawn_async(async move {
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            i * 2
        })
    }).collect();
    
    let results = join_all(futures).await;
    println!("Results: {:?}", results);
}

高级配置

use affinitypool::{AffinityPool, AffinityPoolBuilder};

fn main() {
    let pool = AffinityPoolBuilder::new()
        .num_threads(8)
        .thread_name("worker".to_string())
        .thread_stack_size(2 * 1024 * 1024) // 2MB栈大小
        .affinity(|thread_idx| {
            // 将线程均匀分配到所有核心上
            let core_ids = core_affinity::get_core_ids().unwrap();
            Some(core_ids[thread_idx % core_ids.len()])
        })
        .build()
        .unwrap();
    
    // 使用线程池...
}

完整示例demo

下面是一个完整的示例，展示了如何使用affinitypool进行CPU绑定的矩阵乘法计算：

use affinitypool::AffinityPool;
use core_affinity::CoreId;
use std::sync::Arc;
use std::time::Instant;

const SIZE: usize = 512;

// 矩阵类型
type Matrix = Vec<Vec<f64>>;

fn main() {
    // 初始化两个随机矩阵
    let a = generate_matrix(SIZE);
    let b = generate_matrix(SIZE);
    
    // 获取CPU核心信息
    let core_ids = core_affinity::get_core_ids().unwrap();
    println!("Available CPU cores: {}", core_ids.len());
    
    // 创建线程池，每个线程绑定到不同的核心
    let pool = AffinityPool::with_affinity(core_ids.len(), |thread_idx| {
        Some(core_ids[thread_idx % core_ids.len()])
    }).expect("Failed to create thread pool with affinity");
    
    // 将矩阵包装在Arc中以实现线程安全共享
    let a = Arc::new(a);
    let b = Arc::new(b);
    
    let start = Instant::now();
    
    // 分割工作负载
    let chunk_size = SIZE / core_ids.len();
    let mut handles = Vec::new();
    
    for i in 0..core_ids.len() {
        let a = Arc::clone(&a);
        let b = Arc::clone(&b);
        let start_row = i * chunk_size;
        let end_row = if i == core_ids.len() - 1 {
            SIZE
        } else {
            (i + 1) * chunk_size
        };
        
        // 提交任务到线程池
        handles.push(pool.run(move || {
            let mut result = vec![vec![0.0; SIZE]; end_row - start_row];
            for i in start_row..end_row {
                for j in 0..SIZE {
                    for k in 0..SIZE {
                        result[i - start_row][j] += a[i][k] * b[k][j];
                    }
                }
            }
            (start_row, end_row, result)
        }).unwrap());
    }
    
    // 收集结果
    let mut c = vec![vec![0.0; SIZE]; SIZE];
    for handle in handles {
        let (start, end, partial) = handle.join().unwrap();
        c[start..end].copy_from_slice(&partial);
    }
    
    let duration = start.elapsed();
    println!("Matrix multiplication took: {:?}", duration);
    println!("Sample result: {}", c[0][0]);
}

// 生成随机矩阵
fn generate_matrix(size: usize) -> Matrix {
    (0..size)
        .map(|_| {
            (0..size)
                .map(|_| rand::random::<f64>())
                .collect()
        })
        .collect()
}

性能建议

1. 对于计算密集型任务，将线程数设置为物理核心数
2. 对于I/O密集型任务，可以适当增加线程数
3. 将相关任务绑定到同一个CPU核心上可以提高缓存命中率
4. 避免频繁创建和销毁线程池

注意事项

• 在Linux系统上需要适当的权限来设置CPU亲和性
• 过度使用CPU亲和性可能会影响操作系统的调度优化
• 在某些虚拟化环境中可能无法设置CPU亲和性

affinitypool为需要精细控制线程调度和CPU亲和性的应用提供了强大的工具，特别适合高性能计算、游戏服务器、金融交易等低延迟场景。