Revert "Switch to rayon instead of scoped_threadpool"

This reverts commit bbe2600.

Switching to `rayon` caused a pretty significant performance drop.

Author: rossmacarthur

Cargo.lock

@@ -23,14 +23,14 @@ git2 = "0.13.8"
 globwalk = "0.8.0"
 handlebars = "3.4.0"
 home = "0.5.3"
-indexmap = { version = "1.5.1", features = ["serde-1", "rayon"] }
+indexmap = { version = "1.5.1", features = ["serde-1"] }
 itertools = "0.9.0"
 maplit = "1.0.2"
 once_cell = "1.4.1"
-rayon = "1.3.1"
 regex = "1.3.9"
 regex-macro = "0.1.0"
 reqwest = { version = "0.10.7", features = ["blocking"] }
+scoped_threadpool = "0.1.9"
 serde = { version = "1.0.115", features = ["derive"] }
 structopt = "0.3.16"
 thiserror = "1.0.20"

Cargo.toml

@@ -3,18 +3,20 @@
 //! This module handles the downloading of `Source`s and figuring out which
 //! files to use for `Plugins`.
 
+use std::cmp;
 use std::collections::HashSet;
+use std::convert::TryInto;
 use std::fmt;
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::result;
+use std::sync;
 
 use anyhow::{anyhow, bail, Context as ResultExt, Error, Result};
 use indexmap::{indexmap, IndexMap};
 use itertools::{Either, Itertools};
 use maplit::hashmap;
 use once_cell::sync::Lazy;
-use rayon::prelude::*;
 use serde::{Deserialize, Serialize};
 use url::Url;
 use walkdir::WalkDir;
@@ -547,26 +549,46 @@ impl Config {
                 .map(|(_, locked)| locked)
                 .collect::<Vec<_>>()
         } else {
-            // Install the sources in parallel.
-            map.into_par_iter()
-                .map(|(source, plugins)| {
-                    let source_name = source.to_string();
-                    let source = source
-                        .lock(ctx)
-                        .with_context(s!("failed to install source `{}`", source_name))?;
-
-                    let mut locked = Vec::with_capacity(plugins.len());
-                    for (index, plugin) in plugins {
-                        let name = plugin.name.clone();
-                        locked.push((
-                            index,
-                            plugin
-                                .lock(ctx, &templates, source.clone(), matches, apply)
-                                .with_context(s!("failed to install plugin `{}`", name)),
-                        ));
-                    }
-                    Ok(locked)
-                })
+            /// The maximmum number of threads to use while downloading sources.
+            const MAX_THREADS: u32 = 8;
+
+            // Create a thread pool and install the sources in parallel.
+            let thread_count = cmp::min(count.try_into().unwrap_or(MAX_THREADS), MAX_THREADS);
+            let mut pool = scoped_threadpool::Pool::new(thread_count);
+            let (tx, rx) = sync::mpsc::channel();
+            let templates_ref = &templates;
+
+            pool.scoped(move |scoped| {
+                for (source, plugins) in map {
+                    let tx = tx.clone();
+                    scoped.execute(move || {
+                        tx.send((|| {
+                            let source_name = source.to_string();
+                            let source = source
+                                .lock(ctx)
+                                .with_context(s!("failed to install source `{}`", source_name))?;
+
+                            let mut locked = Vec::with_capacity(plugins.len());
+                            for (index, plugin) in plugins {
+                                let name = plugin.name.clone();
+                                locked.push((
+                                    index,
+                                    plugin
+                                        .lock(ctx, templates_ref, source.clone(), matches, apply)
+                                        .with_context(s!("failed to install plugin `{}`", name)),
+                                ));
+                            }
+                            Ok(locked)
+                        })())
+                        .expect("oops! did main thread die?");
+                    })
+                }
+                scoped.join_all();
+            });
+
+            rx.iter()
+                // all threads must send a response
+                .take(count)
                 // The result of this is basically an `Iter<Result<Vec<(usize, Result)>, _>>`
                 // The first thing we need to do is to filter out the failures and record the
                 // errors that occurred while installing the source in our `errors` list.