Add option to limit number of concurrent threads.

Cargo.lock

@@ -126,6 +126,7 @@ dependencies = [
  "indicatif",
  "lofty",
  "log",
+ "num_cpus",
  "regex",
  "rusqlite",
  "substring",

Cargo.toml

@@ -25,3 +25,4 @@ substring = "1.4.5"
 ureq = "2.4.0"
 configparser = "3.0.0"
 if_chain = "1.0.2"
+num_cpus = "1.13.0"

ChangeLog

@@ -1,3 +1,7 @@
+0.2.3
+-----
+1. Add option to limit number of concurrent threads.
+
 0.2.2
 -----
 1. Update version of tag reader library.

src/analyse.rs

@@ -9,14 +9,16 @@
 use crate::db;
 use crate::tags;
 use anyhow::Result;
-use bliss_audio::{analyze_paths};
+use bliss_audio::{analyze_paths_with_cores};
 use if_chain::if_chain;
 use indicatif::{ProgressBar, ProgressStyle};
 use std::collections::HashSet;
 use std::convert::TryInto;
 use std::fs::{DirEntry, File};
 use std::io::{BufRead, BufReader};
+use std::num::NonZeroUsize;
 use std::path::{Path, PathBuf};
+use num_cpus;
 
 const DONT_ANALYSE: &str = ".notmusic";
 const MAX_ERRORS_TO_SHOW: usize = 100;
@@ -81,13 +83,17 @@ fn check_dir_entry(db: &mut db::Db, mpath: &Path, entry: DirEntry, track_paths:
     }
 }
 
-pub fn analyse_new_files(db: &db::Db, mpath: &PathBuf, track_paths: Vec<String>) -> Result<()> {
+pub fn analyse_new_files(db: &db::Db, mpath: &PathBuf, track_paths: Vec<String>, max_threads: usize) -> Result<()> {
     let total = track_paths.len();
     let progress = ProgressBar::new(total.try_into().unwrap()).with_style(
         ProgressStyle::default_bar()
             .template("[{elapsed_precise}] [{bar:25}] {percent:>3}% {pos:>6}/{len:6} {wide_msg}")
             .progress_chars("=> "),
     );
+    let cpu_threads: NonZeroUsize = match max_threads {
+        0 => NonZeroUsize::new(num_cpus::get()).unwrap(),
+        _ => NonZeroUsize::new(max_threads).unwrap(),
+    };
 
     let mut analysed = 0;
     let mut failed: Vec<String> = Vec::new();
@@ -95,7 +101,7 @@ pub fn analyse_new_files(db: &db::Db, mpath: &PathBuf, track_paths: Vec<String>)
     let mut reported_cue:HashSet<String> = HashSet::new();
 
     log::info!("Analysing new files");
-    for (path, result) in analyze_paths(track_paths) {
+    for (path, result) in analyze_paths_with_cores(track_paths, cpu_threads) {
         let stripped = path.strip_prefix(mpath).unwrap();
         let spbuff = stripped.to_path_buf();
         let sname = String::from(spbuff.to_string_lossy());
@@ -194,7 +200,7 @@ pub fn analyse_new_files(db: &db::Db, mpath: &PathBuf, track_paths: Vec<String>)
     Ok(())
 }
 
-pub fn analyse_files(db_path: &str, mpaths: &Vec<PathBuf>, dry_run: bool, keep_old: bool, max_num_tracks: usize) {
+pub fn analyse_files(db_path: &str, mpaths: &Vec<PathBuf>, dry_run: bool, keep_old: bool, max_num_tracks: usize, max_threads: usize) {
     let mut db = db::Db::new(&String::from(db_path));
     let mut track_count_left = max_num_tracks;
 
@@ -235,7 +241,7 @@ pub fn analyse_files(db_path: &str, mpaths: &Vec<PathBuf>, dry_run: bool, keep_o
             }
 
             if !track_paths.is_empty() {
-                match analyse_new_files(&db, &mpath, track_paths) {
+                match analyse_new_files(&db, &mpath, track_paths, max_threads) {
                     Ok(_) => { }
                     Err(e) => { log::error!("Analysis returned error: {}", e); }
                 }

src/main.rs

@@ -34,6 +34,7 @@ fn main() {
     let mut lms_host = "127.0.0.1".to_string();
     let mut max_num_files: usize = 0;
     let mut music_paths: Vec<PathBuf> = Vec::new();
+    let mut max_threads: usize = 0;
 
     match dirs::home_dir() {
         Some(path) => {
@@ -64,6 +65,7 @@ fn main() {
         arg_parse.refer(&mut ignore_file).add_option(&["-i", "--ignore"], Store, &ignore_file_help);
         arg_parse.refer(&mut lms_host).add_option(&["-L", "--lms"], Store, &lms_host_help);
         arg_parse.refer(&mut max_num_files).add_option(&["-n", "--numfiles"], Store, "Maximum number of files to analyse");
+        arg_parse.refer(&mut max_threads).add_option(&["-t", "--threads"], Store, "Maximum number of threads to use for analysis");
         arg_parse.refer(&mut task).add_argument("task", Store, "Task to perform; analyse, tags, ignore, upload, stopmixer.");
         arg_parse.parse_args_or_exit();
     }
@@ -175,7 +177,7 @@ fn main() {
                 }
                 analyse::update_ignore(&db_path, &ignore_path);
             } else {
-                analyse::analyse_files(&db_path, &music_paths, dry_run, keep_old, max_num_files);
+                analyse::analyse_files(&db_path, &music_paths, dry_run, keep_old, max_num_files, max_threads);
             }
         }
     }