Multiple rusty Sieves of Eratosthenes

Question

To get more familiar with the multi-threading aspects in the Rust language I decided to multi-thread my earlier implementation of The rusty Sieve of Eratosthenes.

I have to say that it is probably in a worse state than the code in that question and it has been written a couple of days ago already, it just wasn't put up yet because the linked question hadn't been reviewed yet.

The implementation uses some techniques of this paper, unfortunately I have not really been able to win speed over the single-threaded version.

It computes the prime numbers by calculating all primes smaller than the square root of the max prime you are searching for on one thread and those results are sent to the other threads and used as divisors. Lastly the threads send back the primes they have calculated and the main thread combines them again.

I am aware of the following shortcomings but would nevertheless want to receive feedback on those:

• The code is not really split over multiple types or files.
• Every thread manages their own piece of memory to store the found primes. *
• The code that prints every prime it has found it explicitly commented out but it is also ready for review, it only takes a while before it has printed out all primes smaller or equal to 5 billion.

(*) After searching I figured out that the split_at_mut method from Vec seems a reasonable candidate, however the bit-vec library does not support that method, so currently I am in the process of writing my own bit vector library to make it possible that only one big chunk of memory is used and that the overhead of combining the results can be avoided.

The code:

extern crate bit_vec;
extern crate nalgebra;
extern crate stopwatch;

use std::thread;
use std::sync::Arc;
use std::sync::mpsc;

use bit_vec::BitVec;
use stopwatch::Stopwatch;

const MAXIMUM_PRIME: usize = 5000000000;
const THREAD_COUNT: usize = 8;

fn main() {
    let sqrt_prime = ((MAXIMUM_PRIME as f64).sqrt() as usize) + 1;
    let actual_thread_count = std::cmp::min(THREAD_COUNT, sqrt_prime);

    println!("Finding primes up to {} on {} threads", MAXIMUM_PRIME, actual_thread_count);

    let stopwatch = Stopwatch::start_new();

    let mut thread_handles = Vec::with_capacity(THREAD_COUNT - 1);
    let mut prime_transmitters = Vec::with_capacity(THREAD_COUNT - 1);

    let (tx_results, rx_results) = mpsc::channel();

    //set up threads
    for tid in 0..(THREAD_COUNT - 1) {
        let block_data = BlockData::create_for_thread(tid + 1, actual_thread_count);

        let (tx_primes, rx_primes) = mpsc::channel();
        prime_transmitters.push(tx_primes);

        let tx_results = tx_results.clone();

        let thread_handle = thread::spawn(move || {
            let mut local_primes = BitVec::from_elem(block_data.size(), true);

            for prime in rx_primes.iter() {
                let mut m = block_data.first_possible_prime_with_divisor(prime);
                while m <= block_data.end_number() {
                    local_primes.set(block_data.to_local_index(m), false);
                    m += prime;
                }
            }

            println!("TID {} finished after {:.3}s", tid, stopwatch.elapsed_ms() as f64 / 1000f64);

            //send found primes back to main
            tx_results.send(ThreadResult {
                tid: tid,
                primes: Arc::new(local_primes)
            }).unwrap();
        });
        thread_handles.push(thread_handle);
    }

    //set up own bit vector
    let first_block_data = BlockData::create_for_thread(0, actual_thread_count);
    let mut primes = BitVec::from_elem(MAXIMUM_PRIME, true);

    primes.set(first_block_data.to_local_index(1), false);

    //find primes on main thread
    for (i, n) in first_block_data.local_index_number_iter() {
        if primes[i] {
            let mut m = n * n;
            while m <= first_block_data.end_number() {
                primes.set(first_block_data.to_local_index(m), false);
                m += n;
            }

            //send primes
            if n >= sqrt_prime {
                continue
            }

            for tx_primes in prime_transmitters.iter() {
                tx_primes.send(n).unwrap();
            }
        }
    }

    println!("Main thread finished after {:.3}s", stopwatch.elapsed_ms() as f64 / 1000f64);

    //close channel to threads
    for tx_primes in prime_transmitters.into_iter() {
        drop(tx_primes);
    }

    for _ in 0..(THREAD_COUNT-1) {
        let result = rx_results.recv().unwrap();

        let result_block_data = BlockData::create_for_thread(result.tid + 1, actual_thread_count);

        for i in 0..result_block_data.size() {
            primes.set(result_block_data.to_global_index(i), result.primes[i]);
        }
    }

    //wait for threads to finish
    for thread_handle in thread_handles.into_iter() {
        thread_handle.join().unwrap();
    }

    //find primes
    //let found_primes: Vec<_> = primes.iter().enumerate().filter(|t| t.1).map(|t| t.0 + 1).collect();
    //println!("Found primes:\n{:?}", found_primes);

    let primes_count = primes.iter().filter(|x| *x).count();
    println!("Number of primes: {}", primes_count);

    println!("Time elapsed: {:.3}s", stopwatch.elapsed_ms() as f64 / 1000f64);
}

#[derive(Debug)]
struct ThreadResult {
    tid: usize,
    primes: Arc<BitVec>
}

#[derive(Debug)]
struct BlockData {
    start_index: usize,
    end_index: usize
}

impl BlockData {
    fn create_for_thread(pid: usize, thread_count: usize) -> BlockData {
        let start_index = ((pid as f32) * (MAXIMUM_PRIME as f32 / thread_count as f32)).floor() as usize;
        let end_index = ((((pid + 1) as f32) * (MAXIMUM_PRIME as f32 / thread_count as f32)).floor() as usize) - 1;
        BlockData { start_index: start_index, end_index: end_index }
    }

    fn size(&self) -> usize {
        self.end_index - self.start_index + 1
    }

    fn to_number(&self, index: usize) -> usize {
        index + self.start_index + 1
    }

    fn to_local_index(&self, number: usize) -> usize {
        number - self.start_index - 1
    }

    fn to_global_index(&self, index: usize) -> usize {
        index + self.start_index
    }

    fn local_index_number_iter<'a>(&'a self) -> Box<Iterator<Item=(usize,usize)> + 'a> {
        let index_iter = 0..self.size();
        let number_iter = (0..self.size()).map(move |i| self.to_number(i));
        Box::new(index_iter.zip(number_iter))
    }

    fn start_number(&self) -> usize {
        self.to_number(0)
    }

    fn end_number(&self) -> usize {
        self.to_number(self.size() - 1)
    }

    fn first_possible_prime_with_divisor(&self, divisor: usize) -> usize {
        let possible_prime = divisor * divisor;
        if possible_prime < self.start_number() {
            if self.start_number() % divisor == 0 {
                return self.start_number();
            }
            return self.start_number() + (divisor - (self.start_number() % divisor));
        }
        possible_prime
    }
}