Cargo.toml

[package]
name = "sieve_multithreaded"
version = "0.1.0"
authors = ["skiwi <email@email.com>"]

[dependencies]
bit-vec = "0.4.3"
nalgebra = "0.8.2"

main.rs

extern crate bit_vec;
extern crate nalgebra;

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

use bit_vec::BitVec;

const MAXIMUM_PRIME: usize = 100;
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 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 tx = tx.clone();
        let block_data = BlockData::create_for_thread(tid + 1, actual_thread_count);
        println!("Block data: (start index = {}, end index = {})", block_data.start_index, block_data.end_index);

        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() {
                println!("TID {} received prime {}", tid, prime);

                let mut m = block_data.first_possible_prime_with_divisor(prime);
                println!("First possible prime: {}", m);
                while m <= block_data.end_number() {
                    println!("TID {} set number {} to no prime (local index = {}, block start index = {}, block end index = {}, divisor = {})", tid, m, block_data.to_local_index(m), block_data.start_index, block_data.end_index, prime);
                    local_primes.set(block_data.to_local_index(m), false);
                    m += prime;
                }
            }

            //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);
    println!("Block data: (start index = {}, end index = {})", first_block_data.start_index, first_block_data.end_index);
    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;
            }

            println!("Found prime: {}", n);

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

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

    for _ in 0..(THREAD_COUNT-1) {
        println!("Attempting to receive result");
        let result = rx_results.recv().unwrap();
        println!("Received result: {:?}", result);

        let result_block_data = BlockData::create_for_thread(result.tid + 1, actual_thread_count);
        // //let result_primes = Arc::try_unwrap(result.primes).unwrap();
        // let result_primes = result.primes;

        for (i, n) in result_block_data.local_index_number_iter() {
            println!("Index {} / Number {}", i, n);
            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);
}

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

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 = (self.start_index..self.end_index).map(move |i| self.to_local_index(self.to_number(i)));
        let number_iter = (self.start_index..self.end_index).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
    }
}