off_thread_iter.rs

#![feature(test)]

extern crate test;

mod offthread {
    use std::sync::mpsc;
    use std::thread;

    pub trait OffThreadIterator: Iterator + Send + Sized + 'static {
        fn off_thread(self) -> mpsc::IntoIter<Self::Item>
            where Self::Item: Send
        {
            let (sender, receiver) = mpsc::channel();

            // Move the iterator to a new thread and run it there.
            thread::spawn(move || {
                for item in self {
                    if sender.send(item).is_err() {
                        // A send error means the receiver has been dropped,
                        // so no further items will be received. Quietly exit.
                        break;
                    }
                }
            });

            // Return an iterator over items received from the worker thread.
            receiver.into_iter()
        }
    }

    impl<I: Iterator + Send + Sized + 'static> OffThreadIterator for I
        where I::Item: Send
    {}
}


// *****************************************************************************
// Benchmarks

// Very inefficient prime testing.
fn is_prime(i: u64) -> bool {
    if i < 2 {
        return false;
    }
    for j in 2 .. i {
        if i % j == 0 {
            return false;
        }
        if j * j > i {
            break;
        }
    }
    true
}

use test::Bencher;

#[bench]
fn bench_it_single_threaded(bencher: &mut Bencher) {
    bencher.iter(|| {
        let a = 20000 .. 30000;
        a.map(|i| (i, is_prime(i)))
            .filter(|&(i, prime)| prime != is_prime(i + 2))
            .count()
    });
}

#[bench]
fn bench_it_multithreaded(bencher: &mut Bencher) {
    use offthread::OffThreadIterator;

    // The same code, with an `.off_thread()` call added.
    bencher.iter(|| {
        let a = 20000 .. 30000;
        a.map(|i| (i, is_prime(i)))
            .off_thread()
            .filter(|&(i, prime)| prime != is_prime(i + 2))
            .count()
    });
}