如何让我的 rust 代码并行运行得更快？

Question

#![feature(map_first_last)]
use num_cpus;

use std::collections::BTreeMap;
use ordered_float::OrderedFloat;

use std::sync::{Arc, Mutex};
use std::thread;


use std::time::Instant;

const MONO_FREQ: [f64; 26] = [
    8.55, 1.60, 3.16, 3.87, 12.1, 2.18, 2.09, 4.96, 7.33, 0.22, 0.81, 4.21, 2.53, 7.17, 7.47, 2.07,
    0.10, 6.33, 6.73, 8.94, 2.68, 1.06, 1.83, 0.19, 1.72, 0.11,
];

fn main() {
    let ciphertext : String = "helloworldthisisatest".to_string();
    concurrent( &ciphertext);
    parallel( &ciphertext);

}

fn concurrent(ciphertext : &String) {
    let start = Instant::now();

    for _ in 0..50000 {

        let mut best_fit : f64 = chi_squared(&ciphertext);
        let mut best_key : u8 = 0;

        for i in 1..26 {
            let test_fit = chi_squared(&decrypt(&ciphertext, i));
            if test_fit < best_fit {
                best_key = i;
                best_fit = test_fit;
            }
        }
    }

    let elapsed = start.elapsed();
    println!("Concurrent : {} ms", elapsed.as_millis());
}

fn parallel(ciphertext : &String) {
    let cpus = num_cpus::get() as u8;
    let start = Instant::now();

    for _ in 0..50000 {

        let mut best_result : f64 = chi_squared(&ciphertext);

        for i in (0..26).step_by(cpus.into()) {
            let results = Arc::new(Mutex::new(BTreeMap::new()));
            let mut threads = vec![];

            for ii in i..i+cpus {
                threads.push(thread::spawn({
                    let clone = Arc::clone(&results);
                    let test = OrderedFloat(chi_squared(&decrypt(&ciphertext, ii)));
                    move || {
                        let mut v = clone.lock().unwrap();
                        v.insert(test, ii);
                    }
                }));
            }
            for t in threads {
                t.join().unwrap();
            }
            let lock = Arc::try_unwrap(results).expect("Lock still has multiple owners");
            let hold = lock.into_inner().expect("Mutex cannot be locked");
            if hold.last_key_value().unwrap().0.into_inner() > best_result {
                best_result = hold.last_key_value().unwrap().0.into_inner();
            }
        }

    }
    let elapsed = start.elapsed();
    println!("Parallel : {} ms", elapsed.as_millis());
}

fn decrypt(ciphertext : &String, shift : u8) -> String {
    ciphertext.chars().map(|x| ((x as u8 + shift - 97) % 26 + 97) as char).collect()
}

pub fn chi_squared(text: &str) -> f64 {     
    let mut result: f64 = 0.0;
    for (pos, i) in get_letter_counts(text).iter().enumerate() {
        let expected = MONO_FREQ[pos] * text.len() as f64 / 100.0;
        result += (*i as f64 - expected).powf(2.0) / expected;
    }
    return result;
}

fn get_letter_counts(text: &str) -> [u64; 26] {
    let mut results: [u64; 26] = [0; 26];
    for i in text.chars() {
        results[((i as u64) - 97) as usize] += 1;
    }
    return results;
}

很抱歉转储了这么多代码，但我不知道问题出在哪里，无论我尝试什么，并行代码似乎都慢了 100 倍左右。

我认为问题可能出在 chi_squared 函数上，因为我不知道它是否并行运行。

我已经尝试过 arc mutex、rayon 和消息传递，并且在应该加快速度时都减慢了速度。我可以做些什么来加快速度？

Answer 1

您的代码在主线程上计算 chi_squared 函数是正确的版本。

  for ii in i..i + cpus {
        let cp = ciphertext.clone();
        let clone = Arc::clone(&results);
        threads.push(thread::spawn(move || {
            let test = OrderedFloat(chi_squared(&decrypt(&cp, ii)));
            let mut v = clone.lock().unwrap();
            v.insert(test, ii);
        }));
}

请注意，是否并行计算并不重要，因为产生 50000*26 线程和线程之间的同步开销首先构成了 100 倍的差异。使用threadpool 实现将减少开销，但结果仍将比单线程版本慢得多。你唯一能做的就是在外循环（0..50000）中分配工作，但我猜你正试图在主循环内进行并行化。