lockfree_job_runner.rs

mod parallel {
    use std::{sync, thread};
    use std::sync::atomic::Ordering::Relaxed;

    pub fn run<F, I, O>(jobs: Vec<I>, num_workers: usize, f: F) -> Vec<O>
        where F: Fn(usize, &I) -> O,
              F: Send + Sync + 'static,
              I: Send + Sync + 'static,
              O: Send + Sync + 'static
    {
        let jobs = sync::Arc::new((sync::atomic::ATOMIC_USIZE_INIT, jobs, f));
        (0..num_workers)
            .map(|worker_id| {
                let jobs = jobs.clone();
                thread::spawn(move || {
                    let mut results = Vec::new();
                    let &(ref index, ref jobs, ref callback) = &*jobs;
                    while let Some(job) = jobs.get(index.fetch_add(1, Relaxed)) {
                        results.push(callback(worker_id, job))
                    }
                    results
                })
            })
            .collect::<Vec<thread::JoinHandle<Vec<O>>>>()
            .into_iter()
            .flat_map(|thr| thr.join().ok().unwrap_or_default())
            .collect()
    }
}

fn main() {
    use std::time;
    let jobs = vec![12, 30, 9, 12, 13, 20, 23, 5, 12, 3, 13, 16, 4, 5, 13, 30, 6, 27, 20, 18];
    let start_time = time::Instant::now();
    let results = parallel::run(jobs, 4, |worker_id, sleep_dur| {
        use std::thread;
        let indent = 11 * worker_id;
        println!("{:indent$}sleep{:3}ms", "", sleep_dur, indent = indent);
        thread::sleep(time::Duration::from_millis(*sleep_dur));
        println!("{:indent$}done", "", indent = indent);
        *sleep_dur
    });
    let elapsed = start_time.elapsed();
    let elapsed_ms = elapsed.as_secs() as u32 * 1000 + elapsed.subsec_nanos() / 1_000_000;
    println!("slept {} ms in {} ms.", results.into_iter().sum::<u64>(), elapsed_ms);
}