add threading for analyze on process command, speeds up heavily when only ffprobe available

src/commands/process.rs

@@ -3,6 +3,7 @@ use std::sync::Mutex;
 use std::thread::scope;
 
 use crate::args::CLIArgs;
+use crate::types::AudioFileInfo;
 use crate::types::File;
 use crate::utils::ascii_reduce::reduce_to_ascii;
 use crate::utils::formats::get_format_handler;
@@ -20,6 +21,8 @@ pub struct ProcessCommandArgs {
 	pub force_replaygain: bool,
 	#[clap(long)]
 	pub replaygain_threads: Option<u32>,
+	#[clap(long)]
+	pub analyze_threads: Option<u32>,
 }
 
 fn rename_file(process_args: &ProcessCommandArgs, file: &mut File) {
@@ -147,6 +150,13 @@ pub fn add_replaygain_tags(file: &File, force: bool) -> Result<(), Box<dyn std::
 	Ok(())
 }
 
+fn analyze_file(file: &File) -> Result<AudioFileInfo, Box<dyn std::error::Error>> {
+	println!("Analysing: {:?}", file.join_path_from_source());
+	let mut handler = get_format_handler(file)?;
+
+	handler.get_audio_file_info(false)
+}
+
 pub fn process_command(
 	_args: CLIArgs,
 	process_args: &ProcessCommandArgs,
@@ -155,12 +165,47 @@ pub fn process_command(
 	let mut files = scan_for_music(&process_args.source)?;
 
 	println!("Analysing Files");
-	for file in files.iter_mut() {
-		println!("Analysing: {:?}", file.join_path_from_source());
+	let threads = process_args.analyze_threads.unwrap_or(0);
 
-		let mut handler = get_format_handler(file)?;
+	if threads <= 1 {
+		for file in files.iter_mut() {
+			let info = analyze_file(file)?;
+			file.info = info;
+		}
+	} else {
+		let jobs: Arc<Mutex<Vec<File>>> = Arc::new(Mutex::new(files.clone()));
+		let new_files: Arc<Mutex<Vec<File>>> = Arc::new(Mutex::new(Vec::new()));
 
-		file.info = handler.get_audio_file_info(false)?;
+		scope(|s| {
+			for i in 0..threads {
+				println!("spawn {}", i);
+
+				s.spawn(|| loop {
+					let mut jobs = jobs.lock().unwrap();
+					let job = jobs.pop();
+					drop(jobs);
+
+					if let Some(job) = job {
+						let result = analyze_file(&job);
+						if let Err(err) = result {
+							panic!("Error analyzing: {}", err)
+						} else {
+							let mut file = job;
+							file.info = result.unwrap();
+							new_files.lock().unwrap().push(file);
+						}
+					} else {
+						break;
+					}
+				});
+			}
+		});
+
+		let new_files = new_files.lock().unwrap();
+		files = Vec::new();
+		for file in new_files.iter() {
+			files.push(file.clone())
+		}
 	}
 
 	println!("Renaming Files");
@@ -171,7 +216,7 @@ pub fn process_command(
 	if !process_args.skip_replaygain && !process_args.dry_run {
 		println!("Adding ReplayGain Tags to Files");
 
-		let threads = process_args.replaygain_threads.unwrap_or(0);
+		let threads = process_args.replaygain_threads.unwrap_or(2);
 
 		if threads <= 1 {
 			for file in files.iter_mut() {
@@ -186,8 +231,9 @@ pub fn process_command(
 				for _ in 0..threads {
 					s.spawn(|| loop {
 						let mut jobs = jobs.lock().unwrap();
-
 						let job = jobs.pop();
+						drop(jobs);
+
 						if let Some(job) = job {
 							let result = add_replaygain_tags(&job, process_args.force_replaygain);
 							if let Err(err) = result {