From 23d491db55f633f071dd5bb304babceb0daf5236 Mon Sep 17 00:00:00 2001 From: CDrummond Date: Fri, 19 Sep 2025 18:19:51 +0100 Subject: [PATCH] Multi-threaded read of existing analysis tags. --- ChangeLog | 1 + src/analyse.rs | 151 +++++++++++++++++++++++++++++++++++++------------ 2 files changed, 115 insertions(+), 37 deletions(-) diff --git a/ChangeLog b/ChangeLog index 18a5a7e..490c344 100644 --- a/ChangeLog +++ b/ChangeLog @@ -4,6 +4,7 @@ 2. Only build symphonia by default. 3. Don't crash if file has no tags. 4. Set log level for tag reader library. +5. Multi-threaded read of existing analysis tags. 0.4.0 ----- diff --git a/src/analyse.rs b/src/analyse.rs index fab5be2..f74eee8 100644 --- a/src/analyse.rs +++ b/src/analyse.rs @@ -23,12 +23,6 @@ use std::fs::DirEntry; use std::num::{NonZero, NonZeroUsize}; use std::path::{Path, PathBuf}; #[cfg(feature = "ffmpeg")] -use std::sync::mpsc; -#[cfg(feature = "ffmpeg")] -use std::sync::mpsc::{Receiver, Sender}; -#[cfg(feature = "ffmpeg")] -use std::thread; -#[cfg(feature = "ffmpeg")] use std::time::Duration; use num_cpus; #[cfg(feature = "libav")] @@ -41,6 +35,9 @@ use bliss_audio::{BlissResult, Song, AnalysisOptions, decoder::Decoder}; use bliss_audio::{AnalysisOptions, decoder::Decoder}; use ureq; use std::time::{SystemTime, UNIX_EPOCH}; +use std::thread; +use std::sync::mpsc; +use std::sync::mpsc::{Receiver, Sender}; const DONT_ANALYSE: &str = ".notmusic"; const MAX_ERRORS_TO_SHOW: usize = 100; @@ -88,7 +85,7 @@ fn send_notif(notifs: &mut NotifInfo, text: &str) { } fn get_file_list(db: &mut db::Db, mpath: &Path, path: &Path, track_paths: &mut Vec, cue_tracks:&mut Vec, file_count:&mut usize, - max_num_files: usize, tagged_file_count:&mut usize, dry_run: bool, notifs: &mut NotifInfo) { + max_num_files: usize, dry_run: bool, notifs: &mut NotifInfo) { if !path.is_dir() { return; } @@ -98,7 +95,7 @@ fn get_file_list(db: &mut db::Db, mpath: &Path, path: &Path, track_paths: &mut V items.sort_by_key(|dir| dir.path()); for item in items { - check_dir_entry(db, mpath, item, track_paths, cue_tracks, file_count, max_num_files, tagged_file_count, dry_run, notifs); + check_dir_entry(db, mpath, item, track_paths, cue_tracks, file_count, max_num_files, dry_run, notifs); if max_num_files>0 && *file_count>=max_num_files { break; } @@ -106,14 +103,14 @@ fn get_file_list(db: &mut db::Db, mpath: &Path, path: &Path, track_paths: &mut V } fn check_dir_entry(db: &mut db::Db, mpath: &Path, entry: DirEntry, track_paths: &mut Vec, cue_tracks:&mut Vec, file_count:&mut usize, - max_num_files: usize, tagged_file_count:&mut usize, dry_run: bool, notifs: &mut NotifInfo) { + max_num_files: usize, dry_run: bool, notifs: &mut NotifInfo) { let pb = entry.path(); if pb.is_dir() { let check = pb.join(DONT_ANALYSE); if check.exists() { log::info!("Skipping '{}', found '{}'", pb.to_string_lossy(), DONT_ANALYSE); } else if max_num_files<=0 || *file_count, tag_error: &mut Vec) { } } +#[derive(Clone, Default, PartialEq)] +pub struct Meta { + pub file: String, + pub meta: db::Metadata +} + +fn read_tags(tracks: Vec, max_threads: usize) -> Receiver { + let mut cpu_threads: usize = match max_threads { + 1111 => num_cpus::get(), + 0 => num_cpus::get(), + _ => max_threads, + }; + if max_threads==1111 && cpu_threads>1 { + cpu_threads -=1; + } + + #[allow(clippy::type_complexity)] + let (tx, rx): ( + Sender, + Receiver, + ) = mpsc::channel(); + if tracks.is_empty() { + return rx; + } + + let mut handles = Vec::new(); + let mut chunk_length = tracks.len() / cpu_threads; + if chunk_length == 0 { + chunk_length = tracks.len(); + } else if chunk_length == 1 && tracks.len() > cpu_threads { + chunk_length = 2; + } + + for chunk in tracks.chunks(chunk_length) { + let tx_thread = tx.clone(); + let owned_chunk = chunk.to_owned(); + let child = thread::spawn(move || { + for track in owned_chunk { + let _ = tx_thread.send(Meta { + file:track.clone(), + meta: tags::read(&track, true), + }); + } + }); + handles.push(child); + } + + rx +} + +fn check_for_tags(db: &db::Db, mpath: &PathBuf, track_paths: Vec, max_threads: usize, notifs: &mut NotifInfo) -> Vec { + let mut untagged_paths:Vec = Vec::new(); + let total = track_paths.len(); + let results = read_tags(track_paths, max_threads); + 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("=> "), + ); + + log::info!("Reading any existing analysis tags"); + send_notif(notifs, "Reading any existing analysis tags"); + for res in results { + let path = PathBuf::from(&res.file); + let stripped = path.strip_prefix(mpath).unwrap(); + let spbuff = stripped.to_path_buf(); + let sname = String::from(spbuff.to_string_lossy()); + progress.set_message(format!("{}", sname)); + if !res.meta.is_empty() && !res.meta.analysis.is_none() { + db.add_track(&sname, &res.meta.clone(), &res.meta.analysis.unwrap()); + } else { + untagged_paths.push(res.file); + } + if terminate_analysis() { + break + } + progress.inc(1); + if notifs.enabled { + let pc = (progress.position() as f64 * 100.0)/total as f64; + send_notif(notifs, &format!("{:8.2}% Read {}", pc, sname)); + } + } + if terminate_analysis() { + progress.abandon_with_message("Terminated!"); + } else { + progress.finish_with_message("Finished!"); + } + untagged_paths +} + #[cfg(not(feature = "ffmpeg"))] fn analyse_new_files(db: &db::Db, mpath: &PathBuf, track_paths: Vec, max_threads: usize, write_tags: bool, preserve_mod_times: bool, notifs: &mut NotifInfo) -> Result<()> { @@ -304,7 +379,7 @@ fn analyse_new_files(db: &db::Db, mpath: &PathBuf, track_paths: Vec, max progress.inc(1); if notifs.enabled { let pc = (progress.position() as f64 * 100.0)/total as f64; - send_notif(notifs, &format!("{:8.2}% {}", pc, sname)); + send_notif(notifs, &format!("{:8.2}% Analysed {}", pc, sname)); } } if terminate_analysis() { @@ -522,24 +597,21 @@ pub fn analyse_files(db_path: &str, mpaths: &Vec, dry_run: bool, keep_o let mut track_paths: Vec = Vec::new(); let mut cue_tracks:Vec = Vec::new(); let mut file_count:usize = 0; - let mut tagged_file_count:usize = 0; log::info!("Looking for new files in {}", mpath.to_string_lossy()); send_notif(&mut notifs, &format!("Looking for new files in {}", mpath.to_string_lossy())); - get_file_list(&mut db, &mpath, &cur, &mut track_paths, &mut cue_tracks, &mut file_count, max_num_files, - &mut tagged_file_count, dry_run, &mut notifs); + get_file_list(&mut db, &mpath, &cur, &mut track_paths, &mut cue_tracks, &mut file_count, max_num_files, dry_run, &mut notifs); track_paths.sort(); - log::info!("New untagged files: {}", track_paths.len()); + log::info!("New files: {}", track_paths.len()); if !cue_tracks.is_empty() { log::info!("New cue tracks: {}", cue_tracks.len()); } - log::info!("New tagged files: {}", tagged_file_count); if !terminate_analysis() { if dry_run { if !track_paths.is_empty() || !cue_tracks.is_empty() { - log::info!("The following need to be analysed:"); + log::info!("The following need to be analysed (or tags read):"); for track in track_paths { log::info!(" {}", track); } @@ -549,13 +621,18 @@ pub fn analyse_files(db_path: &str, mpaths: &Vec, dry_run: bool, keep_o } } else { if !track_paths.is_empty() { - match analyse_new_files(&db, &mpath, track_paths, max_threads, write_tags, preserve_mod_times, &mut notifs) { - Ok(_) => { changes_made = true; } - Err(e) => { log::error!("Analysis returned error: {}", e); } + let untagged_paths = check_for_tags(&db, &mpath, track_paths, max_threads, &mut notifs); + + if !untagged_paths.is_empty() { + log::info!("New untagged files: {}", untagged_paths.len()); + match analyse_new_files(&db, &mpath, untagged_paths, max_threads, write_tags, preserve_mod_times, &mut notifs) { + Ok(_) => { changes_made = true; } + Err(e) => { log::error!("Analysis returned error: {}", e); } + } + } else { + log::info!("No new untagged files to analyse"); + send_notif(&mut notifs, "No new files to analyse"); } - } else { - log::info!("No new files to analyse"); - send_notif(&mut notifs, "No new files to analyse"); } #[cfg(feature = "ffmpeg")]