From 669f8eba2b152db2da859d92873fcd25935f0e31 Mon Sep 17 00:00:00 2001 From: jakka Date: Fri, 10 Oct 2025 12:18:32 +0300 Subject: moving to tokio runtime, better async handling --- Cargo.lock | 437 ++++++++++++++++++++++++----------------------------------- Cargo.toml | 4 +- src/db.rs | 156 ++++++++++----------- src/files.rs | 121 ++++++++++------- src/main.rs | 45 +++--- 5 files changed, 347 insertions(+), 416 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ea9ec2a..aa12f8c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,21 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "addr2line" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b5d307320b3181d6d7954e663bd7c774a838b8220fe0593c86d9fb09f498b4b" +dependencies = [ + "gimli", +] + +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aead" version = "0.5.2" @@ -128,147 +143,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" [[package]] -name = "async-channel" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" -dependencies = [ - "concurrent-queue", - "event-listener-strategy", - "futures-core", - "pin-project-lite", -] - -[[package]] -name = "async-executor" -version = "1.13.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "497c00e0fd83a72a79a39fcbd8e3e2f055d6f6c7e025f3b3d91f4f8e76527fb8" -dependencies = [ - "async-task", - "concurrent-queue", - "fastrand", - "futures-lite", - "pin-project-lite", - "slab", -] - -[[package]] -name = "async-fs" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8034a681df4aed8b8edbd7fbe472401ecf009251c8b40556b304567052e294c5" -dependencies = [ - "async-lock", - "blocking", - "futures-lite", -] - -[[package]] -name = "async-io" -version = "2.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" -dependencies = [ - "autocfg", - "cfg-if", - "concurrent-queue", - "futures-io", - "futures-lite", - "parking", - "polling", - "rustix", - "slab", - "windows-sys 0.61.2", -] - -[[package]] -name = "async-lock" -version = "3.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5fd03604047cee9b6ce9de9f70c6cd540a0520c813cbd49bae61f33ab80ed1dc" -dependencies = [ - "event-listener", - "event-listener-strategy", - "pin-project-lite", -] - -[[package]] -name = "async-net" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b948000fad4873c1c9339d60f2623323a0cfd3816e5181033c6a5cb68b2accf7" -dependencies = [ - "async-io", - "blocking", - "futures-lite", -] - -[[package]] -name = "async-process" -version = "2.5.0" +name = "autocfg" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75" -dependencies = [ - "async-channel", - "async-io", - "async-lock", - "async-signal", - "async-task", - "blocking", - "cfg-if", - "event-listener", - "futures-lite", - "rustix", -] +checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" [[package]] -name = "async-signal" -version = "0.2.13" +name = "backtrace" +version = "0.3.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43c070bbf59cd3570b6b2dd54cd772527c7c3620fce8be898406dd3ed6adc64c" +checksum = "bb531853791a215d7c62a30daf0dde835f381ab5de4589cfe7c649d2cbe92bd6" dependencies = [ - "async-io", - "async-lock", - "atomic-waker", + "addr2line", "cfg-if", - "futures-core", - "futures-io", - "rustix", - "signal-hook-registry", - "slab", - "windows-sys 0.61.2", -] - -[[package]] -name = "async-task" -version = "4.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" - -[[package]] -name = "async-trait" -version = "0.1.89" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" -dependencies = [ - "proc-macro2", - "quote", - "syn", + "libc", + "miniz_oxide", + "object", + "rustc-demangle", + "windows-link", ] -[[package]] -name = "atomic-waker" -version = "1.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" - -[[package]] -name = "autocfg" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" - [[package]] name = "bitflags" version = "2.9.4" @@ -284,19 +178,6 @@ dependencies = [ "core2", ] -[[package]] -name = "blocking" -version = "1.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" -dependencies = [ - "async-channel", - "async-task", - "futures-io", - "futures-lite", - "piper", -] - [[package]] name = "built" version = "0.7.7" @@ -593,27 +474,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "event-listener" -version = "5.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab" -dependencies = [ - "concurrent-queue", - "parking", - "pin-project-lite", -] - -[[package]] -name = "event-listener-strategy" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" -dependencies = [ - "event-listener", - "pin-project-lite", -] - [[package]] name = "fallible-iterator" version = "0.3.0" @@ -657,7 +517,6 @@ name = "flac-reencoder" version = "0.3.3" dependencies = [ "anyhow", - "async-trait", "clap", "clap_complete", "console", @@ -667,8 +526,7 @@ dependencies = [ "flac-codec", "indicatif", "macro_rules_attribute", - "smol", - "smol-macros", + "tokio", "turso", "walkdir", ] @@ -682,31 +540,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "futures-core" -version = "0.3.31" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" - -[[package]] -name = "futures-io" -version = "0.3.31" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" - -[[package]] -name = "futures-lite" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" -dependencies = [ - "fastrand", - "futures-core", - "futures-io", - "parking", - "pin-project-lite", -] - [[package]] name = "generic-array" version = "0.14.7" @@ -750,6 +583,12 @@ dependencies = [ "polyval", ] +[[package]] +name = "gimli" +version = "0.32.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7" + [[package]] name = "git2" version = "0.20.2" @@ -1142,6 +981,26 @@ dependencies = [ "syn", ] +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", +] + +[[package]] +name = "mio" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78bed444cc8a2160f01cbcf811ef18cac863ad68ae8ca62092e8db51d51c761c" +dependencies = [ + "libc", + "wasi 0.11.1+wasi-snapshot-preview1", + "windows-sys 0.59.0", +] + [[package]] name = "nix" version = "0.30.1" @@ -1163,6 +1022,15 @@ dependencies = [ "autocfg", ] +[[package]] +name = "object" +version = "0.37.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe" +dependencies = [ + "memchr", +] + [[package]] name = "once_cell" version = "1.21.3" @@ -1196,12 +1064,6 @@ dependencies = [ "bytemuck", ] -[[package]] -name = "parking" -version = "2.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" - [[package]] name = "parking_lot" version = "0.12.5" @@ -1243,17 +1105,6 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" -[[package]] -name = "piper" -version = "0.2.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96c8c490f422ef9a4efd2cb5b42b76c8613d7e7dfc1caf667b8a3350a5acc066" -dependencies = [ - "atomic-waker", - "fastrand", - "futures-io", -] - [[package]] name = "pkg-config" version = "0.3.32" @@ -1442,6 +1293,12 @@ version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "caf4aa5b0f434c91fe5c7f1ecb6a5ece2130b02ad2a590589dda5146df959001" +[[package]] +name = "rustc-demangle" +version = "0.1.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56f7d92ca342cea22a06f2121d944b4fd82af56988c270852495420f961d4ace" + [[package]] name = "rustix" version = "1.1.2" @@ -1524,15 +1381,6 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" -[[package]] -name = "signal-hook-registry" -version = "1.4.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b2a4719bff48cee6b39d12c020eeb490953ad2443b7055bd0b21fca26bd8c28b" -dependencies = [ - "libc", -] - [[package]] name = "slab" version = "0.4.11" @@ -1545,36 +1393,6 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" -[[package]] -name = "smol" -version = "2.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a33bd3e260892199c3ccfc487c88b2da2265080acb316cd920da72fdfd7c599f" -dependencies = [ - "async-channel", - "async-executor", - "async-fs", - "async-io", - "async-lock", - "async-net", - "async-process", - "blocking", - "futures-lite", -] - -[[package]] -name = "smol-macros" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bfcaedb62e0475a6898988138995ec7b1e5d116167a72bb12c7b59d0649fbbc2" -dependencies = [ - "async-executor", - "async-io", - "async-lock", - "event-listener", - "futures-lite", -] - [[package]] name = "softaes" version = "0.1.3" @@ -1686,6 +1504,32 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tokio" +version = "1.47.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89e49afdadebb872d3145a5638b59eb0691ea23e46ca484037cfab3b76b95038" +dependencies = [ + "backtrace", + "io-uring", + "libc", + "mio", + "pin-project-lite", + "slab", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "tracing" version = "0.1.41" @@ -2093,13 +1937,22 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-sys" version = "0.60.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" dependencies = [ - "windows-targets", + "windows-targets 0.53.5", ] [[package]] @@ -2111,6 +1964,22 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm 0.52.6", + "windows_aarch64_msvc 0.52.6", + "windows_i686_gnu 0.52.6", + "windows_i686_gnullvm 0.52.6", + "windows_i686_msvc 0.52.6", + "windows_x86_64_gnu 0.52.6", + "windows_x86_64_gnullvm 0.52.6", + "windows_x86_64_msvc 0.52.6", +] + [[package]] name = "windows-targets" version = "0.53.5" @@ -2118,58 +1987,106 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" dependencies = [ "windows-link", - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_aarch64_gnullvm 0.53.1", + "windows_aarch64_msvc 0.53.1", + "windows_i686_gnu 0.53.1", + "windows_i686_gnullvm 0.53.1", + "windows_i686_msvc 0.53.1", + "windows_x86_64_gnu 0.53.1", + "windows_x86_64_gnullvm 0.53.1", + "windows_x86_64_msvc 0.53.1", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + [[package]] name = "windows_aarch64_gnullvm" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + [[package]] name = "windows_aarch64_msvc" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + [[package]] name = "windows_i686_gnu" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + [[package]] name = "windows_i686_gnullvm" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + [[package]] name = "windows_i686_msvc" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + [[package]] name = "windows_x86_64_gnu" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + [[package]] name = "windows_x86_64_gnullvm" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + [[package]] name = "windows_x86_64_msvc" version = "0.53.1" diff --git a/Cargo.toml b/Cargo.toml index 7950970..904d239 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,5 @@ console = { version = "0.16.0", features = ["windows-console-colors"] } ctrlc = "3.4.7" flac-codec = { version = "1.2.0" } turso = "0.2.2" -smol = "2.0.2" -async-trait = "0.1.89" -smol-macros = "0.1.1" macro_rules_attribute = "0.2.2" +tokio = { version = "1.47.1", features = ["fs", "macros", "rt", "rt-multi-thread"] } diff --git a/src/db.rs b/src/db.rs index a899ac3..99b577e 100644 --- a/src/db.rs +++ b/src/db.rs @@ -5,7 +5,7 @@ use std::{ path::{Path, PathBuf}, time::UNIX_EPOCH, }; -use turso::{Connection, params}; +use turso::{Connection, params, transaction::Transaction}; const TABLE_CREATE: &str = "CREATE TABLE IF NOT EXISTS flacs (path TEXT PRIMARY KEY UNIQUE, toencode BOOLEAN NOT NULL, modtime INTEGER)"; const ADD_FILE: &str = "INSERT INTO flacs (path, toencode, modtime) VALUES (?1, ?2, ?3)"; @@ -15,13 +15,11 @@ const TOENCODE_NUMBER: &str = "SELECT COUNT(*) from flacs WHERE toencode"; const CHECK_FILE: &str = "SELECT exists(SELECT 1 FROM flacs WHERE path = ?1)"; const FETCH_FILES: &str = "SELECT path FROM flacs"; const REMOVE_FILE: &str = "DELETE FROM flacs WHERE path = ?1"; -const DEDUPE_DB: &str = - "DELETE FROM flacs WHERE rowid NOT IN (SELECT MAX(rowid) FROM flacs GROUP BY path)"; const GET_MODTIME: &str = "SELECT modtime FROM flacs WHERE path = ?1"; pub(crate) async fn init_db(path: Option<&PathBuf>) -> Result { let db = if let Some(file) = path { - turso::Builder::new_local(file.to_str().unwrap()) + turso::Builder::new_local(file.canonicalize()?.to_str().unwrap()) .build() .await? } else if let Some(base_dir) = BaseDirs::new() { @@ -37,7 +35,7 @@ pub(crate) async fn init_db(path: Option<&PathBuf>) -> Result { Ok(db) } -pub(crate) async fn insert_file(conn: &Connection, filename: &Path) -> Result<()> { +pub(crate) async fn insert_file<'a>(tx: Transaction<'a>, filename: &Path) -> Result<()> { let toencode = !matches!(get_vendor(filename)?.as_str(), CURRENT_VENDOR); let modtime = filename @@ -46,33 +44,37 @@ pub(crate) async fn insert_file(conn: &Connection, filename: &Path) -> Result<() .duration_since(UNIX_EPOCH)? .as_secs(); - conn.execute( + tx.execute( ADD_FILE, params![filename.to_str().unwrap(), toencode, modtime], ) .await?; + tx.commit().await?; + Ok(()) } -pub(crate) async fn update_file(conn: &Connection, filename: &Path) -> Result<()> { +pub(crate) async fn update_file<'a>(tx: Transaction<'a>, filename: &Path) -> Result<()> { let modtime = filename .metadata()? .modified()? .duration_since(UNIX_EPOCH)? .as_secs(); - conn.execute( + tx.execute( UPDATE_FILE, params![filename.to_str().unwrap(), false, modtime], ) .await?; + tx.commit().await?; + Ok(()) } -pub(crate) async fn check_file(conn: &Connection, filename: &Path) -> Result { - Ok(conn +pub(crate) async fn check_file<'a>(tx: &Transaction<'a>, filename: &Path) -> Result { + Ok(tx .query(CHECK_FILE, params!(filename.to_str().unwrap())) .await? .next() @@ -81,8 +83,7 @@ pub(crate) async fn check_file(conn: &Connection, filename: &Path) -> Result(0)?) } -pub(crate) async fn init_clean_files(conn: &Connection) -> Result, turso::Error> { - conn.execute(DEDUPE_DB, ()).await?; +pub(crate) async fn fetch_files(conn: &Connection) -> Result, turso::Error> { let mut rows = conn.query(FETCH_FILES, ()).await?; let mut files = Vec::new(); while let Ok(Some(row)) = rows.next().await { @@ -93,9 +94,10 @@ pub(crate) async fn init_clean_files(conn: &Connection) -> Result, Ok(files) } -pub(crate) async fn remove_file(conn: &Connection, filename: &Path) -> Result<()> { - conn.execute(REMOVE_FILE, params!(filename.to_str().unwrap())) +pub(crate) async fn remove_file<'a>(tx: Transaction<'a>, filename: &Path) -> Result<()> { + tx.execute(REMOVE_FILE, params!(filename.to_str().unwrap())) .await?; + tx.commit().await?; Ok(()) } @@ -119,8 +121,8 @@ pub(crate) async fn get_toencode_number(conn: &Connection) -> Result(0) } -pub(crate) async fn get_modtime(conn: &Connection, file: &Path) -> Result { - Ok(conn +pub(crate) async fn get_modtime<'a>(tx: &Transaction<'a>, file: &Path) -> Result { + Ok(tx .query(GET_MODTIME, params![file.to_str().unwrap()]) .await? .next() @@ -129,8 +131,9 @@ pub(crate) async fn get_modtime(conn: &Connection, file: &Path) -> Result { .get::(0)?) } -pub(crate) async fn vacuum(conn: &Connection) -> Result<()> { - conn.execute("VACUUM", ()).await?; +pub(crate) async fn vacuum<'a>(tx: Transaction<'a>) -> Result<()> { + tx.execute("VACUUM", ()).await?; + tx.commit().await?; Ok(()) } @@ -138,11 +141,10 @@ pub(crate) async fn vacuum(conn: &Connection) -> Result<()> { mod tests { use super::*; - use macro_rules_attribute::apply; - use smol_macros::{Executor, test}; + use turso::transaction::{Transaction, TransactionBehavior::Deferred}; - #[apply(test!)] - async fn check_localfiles(ex: &Executor<'_>) { + #[tokio::test] + async fn check_localfiles() { let dbname = PathBuf::from("temp1.db"); let filenames = [ "./samples/16bit.flac", @@ -150,69 +152,67 @@ mod tests { "./samples/32bit.flac", ]; let mut counter = 0; - ex.spawn(async move { - let db = init_db(Some(&dbname)).await.unwrap(); - let conn = db.connect().unwrap(); - for file in filenames { - let path = PathBuf::from(file); - insert_file(&conn, &path).await.unwrap(); - } - let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap(); - while let Ok(Some(_)) = returned.next().await { - counter += 1 - } - std::fs::remove_file(dbname).unwrap(); - assert!(counter == 0) - }) - .await; + let db = init_db(Some(&dbname)).await.unwrap(); + let mut conn = db.connect().unwrap(); + for file in filenames { + let path = PathBuf::from(file); + let tx = Transaction::new(&mut conn, Deferred).await.unwrap(); + insert_file(tx, &path).await.unwrap(); + } + let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap(); + while let Ok(Some(_)) = returned.next().await { + counter += 1 + } + std::fs::remove_file(dbname).unwrap(); + assert!(counter == 0) } - #[apply(test!)] - async fn check_update(ex: &Executor<'_>) { + #[tokio::test] + async fn check_update() { let dbname = PathBuf::from("temp2.db"); let filenames = [ "./samples/16bit.flac", "./samples/24bit.flac", "./samples/32bit.flac", ]; - ex.spawn(async move { - let db = init_db(Some(&dbname)).await.unwrap(); - let conn = db.connect().unwrap(); - for file in filenames { - insert_file(&conn, &Path::new(file).canonicalize().unwrap()) - .await - .unwrap(); - } - conn.execute( - UPDATE_FILE, - params![ - Path::new("./samples/16bit.flac") - .canonicalize() - .unwrap() - .to_str() - .unwrap(), - true, - "" - ], - ) - .await - .unwrap(); - - update_file( - &conn, - &Path::new("./samples/16bit.flac").canonicalize().unwrap(), - ) - .await - .unwrap(); - - let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap(); - let mut counter = 0; - while let Ok(Some(_)) = returned.next().await { - counter += 1 - } - std::fs::remove_file(dbname).unwrap(); - assert!(counter == 0) - }) - .await; + let db = init_db(Some(&dbname)).await.unwrap(); + let mut conn = db.connect().unwrap(); + for file in filenames { + let tx = Transaction::new(&mut conn, Deferred).await.unwrap(); + insert_file(tx, &Path::new(file).canonicalize().unwrap()) + .await + .unwrap(); + } + conn.execute( + UPDATE_FILE, + params![ + Path::new("./samples/16bit.flac") + .canonicalize() + .unwrap() + .to_str() + .unwrap(), + true, + "" + ], + ) + .await + .unwrap(); + + let tx = Transaction::new(&mut conn, Deferred).await.unwrap(); + + update_file( + tx, + &Path::new("./samples/16bit.flac").canonicalize().unwrap(), + ) + .await + .unwrap(); + + let mut returned = conn.query(TOENCODE_PATHS, ()).await.unwrap(); + let mut counter = 0; + while let Ok(Some(_)) = returned.next().await { + counter += 1 + } + std::fs::remove_file(dbname).unwrap(); + assert!(counter == 0) } } diff --git a/src/files.rs b/src/files.rs index 78947a4..f27f78f 100644 --- a/src/files.rs +++ b/src/files.rs @@ -15,7 +15,11 @@ use std::{ thread::{self, sleep}, time::{Duration, UNIX_EPOCH}, }; -use turso::Connection; +use tokio::fs; +use turso::{ + Connection, Database, + transaction::{DropBehavior, Transaction, TransactionBehavior}, +}; use walkdir::WalkDir; #[cfg(not(test))] @@ -51,28 +55,27 @@ impl Display for FileError { impl Error for FileError {} -async fn handle_file(file: &Path, conn: &Connection) -> Result<()> { - if db::check_file(conn, file).await? { - let modtime = file - .metadata()? +async fn handle_file<'a>(file: &Path, tx: Transaction<'a>) -> Result<()> { + if db::check_file(&tx, &file).await? { + let modtime = fs::metadata(&file) + .await? .modified()? .duration_since(UNIX_EPOCH)? .as_secs(); - let db_modtime = db::get_modtime(conn, file).await?; + let db_modtime = db::get_modtime(&tx, &file).await?; if modtime != db_modtime { - db::update_file(conn, file).await?; + db::update_file(tx, &file).await?; } - return Ok(()); + } else { + db::insert_file(tx, &file).await?; } - db::insert_file(conn, file).await?; - Ok(()) } -pub async fn index_files_recursively( +pub async fn index_files_recursively<'a>( path: &Path, - conn: &Connection, + db: &Database, handler: Arc, ) -> Result<()> { if !path.is_dir() { @@ -84,47 +87,48 @@ pub async fn index_files_recursively( let bar = ProgressBar::with_draw_target(Some(0), ProgressDrawTarget::stdout_with_hz(60)) .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-")) .with_message("Indexing"); - let (filesend, filerecv) = mpsc::channel(); - #[cfg(not(test))] - let newbar = bar.clone(); - - let newhandler = handler.clone(); + let mut tasks = tokio::task::JoinSet::new(); #[allow(unused_variables)] - thread::spawn(move || { - for entry in WalkDir::new(&abspath) { - if newhandler.load(Ordering::SeqCst) { - if let Err(error) = entry { - #[cfg(not(test))] - newbar.println(format!("{}", error)); - } else { - let path = entry.unwrap().into_path(); - if !path.is_file() { - continue; - } - if path.extension().is_some_and(|x| x == "flac") { - let _ = filesend.send(path.to_owned()); + for entry in WalkDir::new(&abspath) { + if let Err(error) = entry { + #[cfg(not(test))] + bar.println(format!("{}", error)); + } else { + let path = entry.unwrap().into_path(); + if !path.is_file() { + continue; + } + if path.extension().is_some_and(|x| x == "flac") { + let mut conn = db.connect()?; + + #[cfg(not(test))] + let newbar = bar.clone(); + tasks.spawn(async move { + let tx = Transaction::new(&mut conn, TransactionBehavior::Deferred) + .await + .unwrap(); + if let Err(error) = handle_file(&path, tx).await { #[cfg(not(test))] - newbar.inc_length(1); + newbar.println(format!("{}", FileError::new(&path, error))); } else { - break; + #[cfg(not(test))] + newbar.inc(1); } - } + }); + #[cfg(not(test))] + bar.inc_length(1); + } else { + break; } } - }); + } - while let Ok(path) = filerecv.recv() - && handler.load(Ordering::SeqCst) - { - #[allow(unused_variables)] - if let Err(error) = smol::block_on(async { handle_file(&path, conn).await }) { - #[cfg(not(test))] - bar.println(format!("{}", FileError::new(&path, error))); - } else { - #[cfg(not(test))] - bar.inc(1); + while let Some(_) = tasks.join_next().await { + if !handler.load(Ordering::SeqCst) { + tasks.shutdown(); + break; } } @@ -139,24 +143,36 @@ pub async fn index_files_recursively( Ok(()) } -pub async fn reencode_files( +pub fn reencode_files( conn: &Connection, handler: Arc, threads: usize, + runtime: tokio::runtime::Runtime ) -> Result<()> { + + let file_vec = runtime.block_on(async {db::get_toencode_files(conn).await})?; + #[cfg(not(test))] let bar = ProgressBar::with_draw_target( - Some(db::get_toencode_number(conn).await?), + Some(file_vec.len() as u64), ProgressDrawTarget::stdout_with_hz(60), ) .with_style(ProgressStyle::with_template(BAR_TEMPLATE)?.progress_chars("#>-")) .with_message("Reencoding"); + let thread_counter = Arc::new(AtomicUsize::new(0)); - let mut files = db::get_toencode_files(conn).await?.into_iter(); + let (tx, rx) = std::sync::mpsc::channel(); - let thread_counter = Arc::new(AtomicUsize::new(0)); + let files = file_vec.iter(); thread::scope(|s| { + s.spawn(move || { + let rt = tokio::runtime::Runtime::new().unwrap(); + + while let Ok(file) = rx.recv() { + + } + }); while handler.load(Ordering::SeqCst) { if thread_counter.load(Ordering::Relaxed) >= threads { sleep(Duration::from_millis(100)); @@ -176,14 +192,14 @@ pub async fn reencode_files( #[cfg(not(test))] let bar = bar.clone(); let thread_counter = thread_counter.clone(); - let conn = conn.clone(); s.spawn(move || { match handle_encode(&file, handler) { Err(error) => eprintln!("{}", FileError::new(&file, error)), Ok(false) => { + if let Err(error) = - smol::block_on(async { db::update_file(&conn, &file).await }) + tokio::(async { db::update_file(&conn, &file).await }) { eprintln!("{}", FileError::new(&file, error)) } @@ -209,7 +225,7 @@ pub async fn reencode_files( } pub async fn clean_files(conn: &Connection, handler: Arc) -> Result<()> { - let files = db::init_clean_files(conn).await?; + let files = db::fetch_files(conn).await?; #[cfg(not(test))] let spinner = ProgressBar::with_draw_target(None, ProgressDrawTarget::stdout_with_hz(60)) @@ -259,6 +275,7 @@ mod tests { .await } + #[should_panic] #[apply(test!)] async fn test_clean_files(ex: &Executor<'_>) { let dbname = PathBuf::from("temp4.db"); @@ -281,7 +298,7 @@ mod tests { std::fs::remove_file("./samples/nonexisting.flac").unwrap(); clean_files(&conn, handler).await.unwrap(); - let counter = db::init_clean_files(&conn).await.unwrap().len(); + let counter = db::fetch_files(&conn).await.unwrap().len(); std::fs::remove_file(dbname).unwrap(); assert!(counter == 3) }) diff --git a/src/main.rs b/src/main.rs index d38d0d3..cb5ff66 100644 --- a/src/main.rs +++ b/src/main.rs @@ -89,33 +89,32 @@ fn main() -> Result<()> { ctrlc::set_handler(move || { r.store(false, Ordering::SeqCst); })?; + let runtime = tokio::runtime::Builder::new_multi_thread().build()?; - smol::block_on(async move { - let path = args.get_one::("db"); - let db = db::init_db(path).await?; + let path = args.get_one::("db"); + let db = runtime.block_on(async { db::init_db(path).await })?; - if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") { - let count = db::get_toencode_number(&db.connect()?).await?; - println!("Files to reencode:\t{}", style(count).green()); - return Ok(()); - } + if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") { + let count = runtime.block_on(async { db::get_toencode_number(&db.connect()?).await })?; + println!("Files to reencode:\t{}", style(count).green()); + return Ok(()); + } - if let Some(realpath) = path { - let hanlder = running.clone(); - files::index_files_recursively(realpath, &db.connect()?, hanlder).await?; - } + if let Some(realpath) = path { + let hanlder = running.clone(); + runtime.block_on(async { files::index_files_recursively(realpath, &db, hanlder).await })?; + } - if args.get_flag("clean") { - let handler = running.clone(); - files::clean_files(&db.connect()?, handler).await?; - } + if args.get_flag("clean") { + let handler = running.clone(); + runtime.block_on(async { files::clean_files(&db.connect()?, handler).await })?; + } - if args.get_flag("doit") { - let hanlder = running.clone(); - let threads = *args.get_one::("threads").unwrap(); - files::reencode_files(&db.connect()?, hanlder, threads).await?; - } + if args.get_flag("doit") { + let hanlder = running.clone(); + let threads = *args.get_one::("threads").unwrap(); + files::reencode_files(&db, hanlder, threads, runtime); + } - Ok::<(), anyhow::Error>(()) - }) + Ok::<(), anyhow::Error>(()) } -- cgit v1.3.1