Function run_threaded

crates/openquant/src/hpc_parallel.rs:276–380

(
    atoms: &[A],
    partitions: &[MoleculePartition],
    progress_every: usize,
    num_threads: usize,
    callback: F,
)

}

fn run_threaded<A, R, F, E>(
    atoms: &[A],
    partitions: &[MoleculePartition],
    progress_every: usize,
    num_threads: usize,
    callback: F,
) -> Result<(Vec<R>, Vec<ProgressSnapshot>), HpcParallelError>
where
    A: Sync,
    R: Send,
    F: Fn(&[A]) -> Result<R, E> + Send + Sync,
    E: Display,
{
    let started = Instant::now();
    let total = partitions.len();
    let callback = Arc::new(callback);

    let (job_tx, job_rx) = mpsc::channel::<MoleculePartition>();
    let job_rx = Arc::new(Mutex::new(job_rx));
    let (result_tx, result_rx) =
        mpsc::channel::<(MoleculePartition, Result<R, HpcParallelError>)>();

    thread::scope(|scope| {
        let mut workers = Vec::with_capacity(num_threads);
        for _ in 0..num_threads {
            let rx = Arc::clone(&job_rx);
            let tx = result_tx.clone();
            let cb = Arc::clone(&callback);
            workers.push(scope.spawn(move || loop {
                let next = {
                    let guard = rx.lock().expect("job receiver mutex should not be poisoned");
                    guard.recv()
                };
                let part = match next {
                    Ok(part) => part,
                    Err(_) => break,
                };
                let res = cb(&atoms[part.start..part.end]).map_err(|err| {
                    HpcParallelError::CallbackFailed {
                        molecule_id: part.molecule_id,
                        message: err.to_string(),
                    }
                });
                if tx.send((part, res)).is_err() {
                    break;
                }
            }));
        }
        drop(result_tx);

        for part in partitions {
            if job_tx.send(*part).is_err() {
                return Err(HpcParallelError::ChannelClosed("queueing jobs"));
            }
        }
        drop(job_tx);

        let mut ordered: Vec<Option<R>> = (0..total).map(|_| None).collect();