summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorjakka <jakkadoujin@gmail.com>2025-07-02 11:42:15 +0300
committerjakka <jakkadoujin@gmail.com>2025-07-02 11:42:15 +0300
commit75a1804c370f7b25540744fcd9f1ae89dedaf7ad (patch)
tree42da6f30b80b366d9a3bbff08ef00a87265ba83f
parent8d36fb00058262721ba571649e1e916690854aae (diff)
trying out parallelism
-rw-r--r--Cargo.lock382
-rw-r--r--Cargo.toml14
-rw-r--r--src/db.rs142
-rw-r--r--src/files.rs173
-rw-r--r--src/main.rs89
5 files changed, 584 insertions, 216 deletions
diff --git a/Cargo.lock b/Cargo.lock
index 2eb7f8e..2ca3c7c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -127,6 +127,121 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e16d2d3311acee920a9eb8d33b8cbc1787ce4a264e85f964c2404b969bdcd487"
[[package]]
+name = "async-channel"
+version = "2.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "89b47800b0be77592da0afd425cc03468052844aff33b84e33cc696f64e77b6a"
+dependencies = [
+ "concurrent-queue",
+ "event-listener-strategy",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-executor"
+version = "1.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bb812ffb58524bdd10860d7d974e2f01cc0950c2438a74ee5ec2e2280c6c4ffa"
+dependencies = [
+ "async-task",
+ "concurrent-queue",
+ "fastrand",
+ "futures-lite",
+ "pin-project-lite",
+ "slab",
+]
+
+[[package]]
+name = "async-fs"
+version = "2.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ebcd09b382f40fcd159c2d695175b2ae620ffa5f3bd6f664131efff4e8b9e04a"
+dependencies = [
+ "async-lock",
+ "blocking",
+ "futures-lite",
+]
+
+[[package]]
+name = "async-io"
+version = "2.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1237c0ae75a0f3765f58910ff9cdd0a12eeb39ab2f4c7de23262f337f0aacbb3"
+dependencies = [
+ "async-lock",
+ "cfg-if",
+ "concurrent-queue",
+ "futures-io",
+ "futures-lite",
+ "parking",
+ "polling",
+ "rustix 1.0.7",
+ "slab",
+ "tracing",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
+name = "async-lock"
+version = "3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff6e472cdea888a4bd64f342f09b3f50e1886d32afe8df3d663c01140b811b18"
+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.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cde3f4e40e6021d7acffc90095cbd6dc54cb593903d1de5832f435eb274b85dc"
+dependencies = [
+ "async-channel",
+ "async-io",
+ "async-lock",
+ "async-signal",
+ "async-task",
+ "blocking",
+ "cfg-if",
+ "event-listener",
+ "futures-lite",
+ "rustix 1.0.7",
+ "tracing",
+]
+
+[[package]]
+name = "async-signal"
+version = "0.2.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d7605a4e50d4b06df3898d5a70bf5fde51ed9059b0434b73105193bc27acce0d"
+dependencies = [
+ "async-io",
+ "async-lock",
+ "atomic-waker",
+ "cfg-if",
+ "futures-core",
+ "futures-io",
+ "rustix 1.0.7",
+ "signal-hook-registry",
+ "slab",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
name = "async-stream"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -149,6 +264,12 @@ dependencies = [
]
[[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.88"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -160,6 +281,12 @@ dependencies = [
]
[[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"
@@ -294,6 +421,19 @@ dependencies = [
]
[[package]]
+name = "blocking"
+version = "1.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "703f41c54fc768e63e091340b424302bb1c29ef4aa0c7f10fe849dfb114d29ea"
+dependencies = [
+ "async-channel",
+ "async-task",
+ "futures-io",
+ "futures-lite",
+ "piper",
+]
+
+[[package]]
name = "bumpalo"
version = "3.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -354,6 +494,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9555578bc9e57714c812a1f84e4fc5b4d21fcb063490c624de019f7464c91268"
[[package]]
+name = "cfg_aliases"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
+
+[[package]]
name = "chrono"
version = "0.4.41"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -445,6 +591,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75"
[[package]]
+name = "concurrent-queue"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
+dependencies = [
+ "crossbeam-utils",
+]
+
+[[package]]
name = "console"
version = "0.15.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -505,6 +660,31 @@ dependencies = [
]
[[package]]
+name = "crossbeam-deque"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51"
+dependencies = [
+ "crossbeam-epoch",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-epoch"
+version = "0.9.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+dependencies = [
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
+
+[[package]]
name = "crypto-common"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -515,6 +695,16 @@ dependencies = [
]
[[package]]
+name = "ctrlc"
+version = "3.4.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "46f93780a459b7d656ef7f071fe699c4d3d2cb201c4b24d085b6ddc505276e73"
+dependencies = [
+ "nix",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
name = "digest"
version = "0.10.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -574,6 +764,27 @@ dependencies = [
]
[[package]]
+name = "event-listener"
+version = "5.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3492acde4c3fc54c845eaab3eed8bd00c7a7d881f78bfc801e43a93dec1331ae"
+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.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -592,6 +803,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
[[package]]
+name = "fastrand"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"
+
+[[package]]
name = "flac-bound"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -609,17 +826,20 @@ dependencies = [
"clap_complete",
"claxon",
"console 0.15.11",
+ "ctrlc",
"directories",
"flac-bound",
"futures-util",
"i24",
"indicatif",
"libsql",
+ "macro_rules_attribute",
"md-5",
"metaflac",
"pin-utils",
- "tokio",
- "tokio-util",
+ "rayon",
+ "smol",
+ "smol-macros",
"walkdir",
]
@@ -678,6 +898,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
[[package]]
+name = "futures-lite"
+version = "2.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f5edaec856126859abb19ed65f39e90fea3a9574b9707f13539acf4abf7eb532"
+dependencies = [
+ "fastrand",
+ "futures-core",
+ "futures-io",
+ "parking",
+ "pin-project-lite",
+]
+
+[[package]]
name = "futures-macro"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -814,6 +1047,12 @@ dependencies = [
]
[[package]]
+name = "hermit-abi"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
+
+[[package]]
name = "hex"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1224,6 +1463,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab"
[[package]]
+name = "linux-raw-sys"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd945864f07fe9f5371a27ad7b52a172b4b499999f1d97574c9fa68373937e12"
+
+[[package]]
name = "lock_api"
version = "0.4.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1240,6 +1485,22 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94"
[[package]]
+name = "macro_rules_attribute"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "65049d7923698040cd0b1ddcced9b0eb14dd22c5f86ae59c3740eab64a676520"
+dependencies = [
+ "macro_rules_attribute-proc_macro",
+ "paste",
+]
+
+[[package]]
+name = "macro_rules_attribute-proc_macro"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "670fdfda89751bc4a84ac13eaa63e205cf0fd22b4c9a5fbfa085b63c1f1d3a30"
+
+[[package]]
name = "matchit"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1304,6 +1565,18 @@ dependencies = [
]
[[package]]
+name = "nix"
+version = "0.30.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6"
+dependencies = [
+ "bitflags 2.9.1",
+ "cfg-if",
+ "cfg_aliases",
+ "libc",
+]
+
+[[package]]
name = "nom"
version = "7.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1356,6 +1629,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
[[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.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1379,6 +1658,12 @@ dependencies = [
]
[[package]]
+name = "paste"
+version = "1.0.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
+
+[[package]]
name = "peeking_take_while"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1462,6 +1747,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
[[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 = "polling"
+version = "3.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b53a684391ad002dd6a596ceb6c74fd004fdce75f4be2e3f615068abbea5fd50"
+dependencies = [
+ "cfg-if",
+ "concurrent-queue",
+ "hermit-abi",
+ "pin-project-lite",
+ "rustix 1.0.7",
+ "tracing",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
name = "portable-atomic"
version = "1.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1564,6 +1875,26 @@ dependencies = [
]
[[package]]
+name = "rayon"
+version = "1.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b418a60154510ca1a002a752ca9714984e21e4241e804d32555251faf8b78ffa"
+dependencies = [
+ "either",
+ "rayon-core",
+]
+
+[[package]]
+name = "rayon-core"
+version = "1.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1465873a3dfdaa8ae7cb14b4383657caab0b3e8a0aa9ae8e04b044854c8dfce2"
+dependencies = [
+ "crossbeam-deque",
+ "crossbeam-utils",
+]
+
+[[package]]
name = "redox_syscall"
version = "0.5.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1647,7 +1978,20 @@ dependencies = [
"bitflags 2.9.1",
"errno",
"libc",
- "linux-raw-sys",
+ "linux-raw-sys 0.4.15",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
+name = "rustix"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c71e83d6afe7ff64890ec6b71d6a69bb8a610ab78ce364b3352876bb4c801266"
+dependencies = [
+ "bitflags 2.9.1",
+ "errno",
+ "libc",
+ "linux-raw-sys 0.9.4",
"windows-sys 0.59.0",
]
@@ -1832,6 +2176,36 @@ 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 = "socket2"
version = "0.5.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2319,7 +2693,7 @@ dependencies = [
"either",
"home",
"once_cell",
- "rustix",
+ "rustix 0.38.44",
]
[[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 54ff725..d6e643e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,7 +17,6 @@ directories = "6.0.0"
flac-bound = { version = "0.5.0", features = [
"libflac-noogg",
], default-features = false }
-futures-util = "0.3.31"
i24 = { version = "2.1.0", features = ["std"] }
indicatif = { version = "0.17.11", features = [
"tokio",
@@ -27,14 +26,13 @@ indicatif = { version = "0.17.11", features = [
libsql = { version = "0.9.11" }
md-5 = "0.10.6"
pin-utils = "0.1.0"
-tokio = { version = "1.45.1", features = [
- "macros",
- "rt",
- "rt-multi-thread",
- "signal",
-] }
-tokio-util = "0.7.15"
walkdir = "2.5.0"
console = { version = "0.15.11", features = ["windows-console-colors"] }
claxon = { git = "https://github.com/justjakka/claxon.git" }
metaflac = "0.2.8"
+rayon = "1.10.0"
+smol = "2.0.2"
+futures-util = "0.3.31"
+smol-macros = "0.1.1"
+macro_rules_attribute = "0.2.2"
+ctrlc = "3.4.7"
diff --git a/src/db.rs b/src/db.rs
index dd32171..ea8d5d2 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -1,7 +1,7 @@
use anyhow::{Result, anyhow};
use directories::BaseDirs;
-use futures_util::Stream;
use libsql::{Builder, Connection, params};
+use smol::stream::Stream;
use std::{
path::Path,
time::{Duration, UNIX_EPOCH},
@@ -151,84 +151,90 @@ pub async fn open_default_db() -> Result<Database> {
#[cfg(test)]
mod tests {
use futures_util::StreamExt;
+ use macro_rules_attribute::apply;
+ use smol_macros::{Executor, test};
use super::*;
- #[tokio::test]
- async fn check_localfiles() {
- let dbname = String::from("temp1.db");
- let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"];
- let conn = Database::new(&dbname).await.unwrap();
- for file in filenames {
- let _ = conn.insert_file(&file.to_string()).await;
- }
- let returned = conn
- .0
- .query(TOENCODE_QUERY, ())
- .await
- .unwrap()
- .into_stream();
- pin_utils::pin_mut!(returned);
-
- let mut counter = 0;
-
- while let Some(Ok(_)) = returned.next().await {
- counter += 1
- }
+ #[apply(test!)]
+ async fn check_localfiles(ex: &Executor<'_>) {
+ ex.spawn(async {
+ let dbname = String::from("temp1.db");
+ let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"];
+ let mut counter = 0;
+ let conn = Database::new(&dbname).await.unwrap();
+ for file in filenames {
+ let _ = conn.insert_file(&file.to_string()).await;
+ }
+ let returned = conn
+ .0
+ .query(TOENCODE_QUERY, ())
+ .await
+ .unwrap()
+ .into_stream();
+ pin_utils::pin_mut!(returned);
- std::fs::remove_file(dbname).unwrap();
- assert!(counter == 0)
+ while let Some(Ok(_)) = returned.next().await {
+ counter += 1
+ }
+ std::fs::remove_file(dbname).unwrap();
+ assert!(counter == 0)
+ })
+ .await;
}
- #[tokio::test]
- async fn check_update() {
- let dbname = String::from("temp2.db");
- let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"];
- let conn = Database::new(&dbname).await.unwrap();
- for file in filenames {
+ #[apply(test!)]
+ async fn check_update(ex: &Executor<'_>) {
+ ex.spawn(async {
+ let dbname = String::from("temp2.db");
+ let filenames = ["16bit.flac", "24bit.flac", "32bit.flac"];
+ let conn = Database::new(&dbname).await.unwrap();
+ for file in filenames {
+ let _ = conn
+ .insert_file(Path::new(file).canonicalize().unwrap())
+ .await;
+ }
+
let _ = conn
- .insert_file(Path::new(file).canonicalize().unwrap())
+ .0
+ .execute(
+ REPLACE_ITEM,
+ params![
+ Path::new("16bit.flac")
+ .canonicalize()
+ .unwrap()
+ .to_str()
+ .unwrap(),
+ true,
+ ""
+ ],
+ )
.await;
- }
- let _ = conn
- .0
- .execute(
- REPLACE_ITEM,
- params![
- Path::new("16bit.flac")
- .canonicalize()
- .unwrap()
- .to_str()
- .unwrap(),
- true,
- ""
- ],
+ conn.update_file(
+ Path::new("16bit.flac")
+ .canonicalize()
+ .unwrap()
+ .to_str()
+ .unwrap(),
)
- .await;
+ .await
+ .unwrap();
- conn.update_file(
- Path::new("16bit.flac")
- .canonicalize()
+ let returned = conn
+ .0
+ .query(TOENCODE_QUERY, ())
+ .await
.unwrap()
- .to_str()
- .unwrap(),
- )
- .await
- .unwrap();
-
- let returned = conn
- .0
- .query(TOENCODE_QUERY, ())
- .await
- .unwrap()
- .into_stream();
- pin_utils::pin_mut!(returned);
- let mut counter = 0;
- while let Some(Ok(_)) = returned.next().await {
- counter += 1
- }
- std::fs::remove_file(dbname).unwrap();
- assert!(counter == 0)
+ .into_stream();
+ pin_utils::pin_mut!(returned);
+ let mut counter = 0;
+ while let Some(Ok(_)) = returned.next().await {
+ counter += 1
+ }
+ std::fs::remove_file(dbname).unwrap();
+ assert!(counter == 0)
+ })
+ .await;
}
}
diff --git a/src/files.rs b/src/files.rs
index f9352d3..b23f6a6 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -3,14 +3,22 @@ use futures_util::StreamExt;
#[cfg(not(test))]
use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle};
use pin_utils::pin_mut;
+use rayon::prelude::*;
+use smol::{
+ Executor,
+ fs::{File, metadata},
+};
use std::{
error::Error,
fmt::Display,
path::{Path, PathBuf},
+ sync::{
+ Arc,
+ atomic::{AtomicBool, Ordering},
+ mpsc,
+ },
time::UNIX_EPOCH,
};
-use tokio::task::JoinSet;
-use tokio_util::sync::CancellationToken;
use walkdir::WalkDir;
use crate::{db::Database, flac::handle_encode};
@@ -51,9 +59,8 @@ impl Error for FileError {}
async fn handle_file(file: impl AsRef<Path>, conn: Database) -> Result<()> {
match conn.check_file(&file).await {
Ok(true) => {
- let modtime = file
- .as_ref()
- .metadata()?
+ let modtime = metadata(file.as_ref())
+ .await?
.modified()?
.duration_since(UNIX_EPOCH)?
.as_secs();
@@ -76,61 +83,71 @@ async fn handle_file(file: impl AsRef<Path>, conn: Database) -> Result<()> {
Ok(())
}
-pub async fn index_files_recursively(
+pub fn index_files_recursively(
path: impl AsRef<Path>,
conn: &Database,
- canceltoken: CancellationToken,
+ running: Arc<AtomicBool>,
) -> Result<()> {
if !path.as_ref().is_dir() {
return Err(anyhow!("Invalid root directory"));
}
let abspath = path.as_ref().canonicalize()?;
- let mut tasks: JoinSet<Result<(), anyhow::Error>> = JoinSet::new();
#[cfg(not(test))]
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");
- for entry in WalkDir::new(abspath) {
- let path = entry?.into_path();
- if !path.is_file() {
- continue;
- }
+ let (tx, rx) = mpsc::channel();
+ let ex = Executor::new();
- if path.extension().is_some_and(|x| x == "flac") {
- let newconn = conn.clone();
- let newtoken = canceltoken.clone();
- #[cfg(not(test))]
- let newbar = bar.clone();
+ WalkDir::new(abspath)
+ .into_iter()
+ .par_bridge()
+ .for_each(|entry| {
+ if !running.load(Ordering::SeqCst) {
+ return;
+ } else {
+ let path = entry.unwrap().into_path();
+ if !path.is_file() {
+ return;
+ }
+ if path.extension().is_some_and(|x| x == "flac") {
+ let newconn = conn.clone();
+ let newrunning = running.clone();
+ let newtx = tx.clone();
+ #[cfg(not(test))]
+ let newbar = bar.clone();
- tasks.spawn(async move {
- tokio::select! {
- _ = newtoken.cancelled() => Ok(()),
- res = async {
- handle_file(path, newconn).await?;
- #[cfg(not(test))]
- newbar.inc(1);
- Ok(())
- } => res
+ ex.spawn(async move {
+ if !newrunning.load(Ordering::SeqCst) {
+ match handle_file(&path, newconn).await {
+ Err(error) => newtx.send(FileError::new(path, error)),
+ Ok(_) => {
+ #[cfg(not(test))]
+ newbar.inc(1);
+ Ok(())
+ }
+ }
+ } else {
+ Ok(())
+ }
+ })
+ .detach();
+
+ #[cfg(not(test))]
+ bar.inc_length(1);
}
- });
+ }
+ });
- #[cfg(not(test))]
- bar.inc_length(1);
- }
+ while let Ok(message) = rx.recv() {
+ eprintln!("{}", message);
}
- while let Some(task) = tasks.join_next().await {
- match task {
- Ok(Err(error)) => eprintln!("{error}"),
- Err(error) => eprintln!("Error encountered:\t{}", error),
- _ => {}
- }
- }
#[cfg(not(test))]
{
- if canceltoken.is_cancelled() {
+ if !running.load(Ordering::SeqCst) {
bar.abandon_with_message("Indexing aborted");
} else {
bar.finish_with_message("Finished indexing");
@@ -139,12 +156,10 @@ pub async fn index_files_recursively(
Ok(())
}
-pub async fn reencode_files(conn: &Database, canceltoken: CancellationToken) -> Result<()> {
+/* pub fn reencode_files(conn: &Database) -> Result<()> {
let stream = conn.get_toencode_stream().await?;
pin_mut!(stream);
- let mut tasks = JoinSet::new();
-
#[cfg(not(test))]
let bar = ProgressBar::with_draw_target(
Some(conn.get_toencode_number().await?),
@@ -186,27 +201,12 @@ pub async fn reencode_files(conn: &Database, canceltoken: CancellationToken) ->
}
}
- while let Some(task) = tasks.join_next().await {
- match task {
- Ok(Err(error)) => eprintln!("Error encountered:\t{error}"),
- Err(error) => eprintln!("Error encountered:\t{error}"),
- _ => {}
- }
- }
-
- #[cfg(not(test))]
- {
- if canceltoken.is_cancelled() {
- bar.abandon_with_message("Reencoding aborted");
- } else {
- bar.finish_with_message("Finished encoding");
- }
- }
-
Ok(())
}
-pub async fn clean_files(conn: &Database) -> Result<()> {
+pub fn clean_files(conn: &Database) -> Result<()> {
+ let ex = Executor::new();
+
let mut tasks: JoinSet<std::result::Result<(), anyhow::Error>> = JoinSet::new();
let query_res = conn.init_clean_files().await?;
@@ -237,42 +237,41 @@ pub async fn clean_files(conn: &Database) -> Result<()> {
conn.vaccum().await?;
Ok(())
-}
+} */
#[cfg(test)]
mod tests {
use super::*;
+ use macro_rules_attribute::apply;
+ use smol_macros::{Executor, test};
- #[tokio::test]
- async fn test_index_lots_of_files() {
- let conn = Database::new("temp3.db").await.unwrap();
- let token = CancellationToken::new();
- index_files_recursively(Path::new("./testfiles"), &conn, token)
- .await
- .unwrap();
+ #[apply(test!)]
+ async fn test_index_lots_of_files(ex: &Executor<'_>) {
+ ex.spawn(async {
+ let running = Arc::new(AtomicBool::new(true));
+ let r = running.clone();
- std::fs::remove_file("temp3.db").unwrap();
+ ctrlc::set_handler(move || {
+ r.store(false, Ordering::SeqCst);
+ })
+ .unwrap();
+ let conn = Database::new("temp3.db").await.unwrap();
+ index_files_recursively(Path::new("./testfiles"), &conn, running).unwrap();
+ std::fs::remove_file("temp3.db").unwrap();
+ })
+ .await;
}
- #[test]
- fn test_reencode_lots_of_files() {
- let runtime = tokio::runtime::Builder::new_multi_thread()
- .max_blocking_threads(4)
- .enable_all()
- .build()
- .unwrap();
- runtime.block_on(async move {
+ /* #[apply(test!)]
+ async fn test_reencode_lots_of_files(ex: &Executor<'_>) {
+ ex.spawn(async {
let conn = Database::new("temp4.db").await.unwrap();
- let token = CancellationToken::new();
- index_files_recursively(Path::new("./testfiles"), &conn, token)
- .await
- .unwrap();
+ index_files_recursively(Path::new("./testfiles"), &conn).unwrap();
println!("\n{}", conn.get_toencode_number().await.unwrap());
- let token = CancellationToken::new();
- reencode_files(&conn, token).await.unwrap();
+ reencode_files(&conn).unwrap();
println!("\n{}", conn.get_toencode_number().await.unwrap());
- });
-
- std::fs::remove_file("temp4.db").unwrap();
- }
+ std::fs::remove_file("temp4.db").unwrap();
+ })
+ .await;
+ } */
}
diff --git a/src/main.rs b/src/main.rs
index 9deb7c6..6db57a3 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -5,8 +5,13 @@ use anyhow::Result;
use clap::{Arg, ArgAction, Command, ValueHint, command, value_parser};
use clap_complete::{Generator, Shell, generate};
use console::style;
-use std::path::PathBuf;
-use tokio_util::sync::CancellationToken;
+use std::{
+ path::PathBuf,
+ sync::{
+ Arc,
+ atomic::{AtomicBool, Ordering},
+ },
+};
fn build_cli() -> Command {
command!()
@@ -69,6 +74,13 @@ fn print_completions<G: Generator>(generator: G, cmd: &mut Command) {
}
fn main() -> Result<()> {
+ let running = Arc::new(AtomicBool::new(true));
+ let r = running.clone();
+
+ ctrlc::set_handler(move || {
+ r.store(false, Ordering::SeqCst);
+ })?;
+
let args = build_cli().get_matches();
if let Some(generator) = args.get_one::<Shell>("shell").copied() {
@@ -78,57 +90,36 @@ fn main() -> Result<()> {
return Ok(());
}
- let runtime = tokio::runtime::Builder::new_multi_thread()
- .max_blocking_threads(*args.get_one::<usize>("threads").unwrap())
- .enable_all()
- .build()?;
- runtime.block_on(async move {
- let conn = if let Some(path) = args.get_one::<PathBuf>("db") {
- db::Database::new(path).await?
- } else {
- db::open_default_db().await?
- };
-
- let root_canceltoken = CancellationToken::new();
-
- let canceltoken = root_canceltoken.clone();
-
- tokio::spawn(async move {
- let _ = tokio::signal::ctrl_c().await;
- root_canceltoken.cancel();
- });
-
- let path = args.get_one::<PathBuf>("path");
+ let conn = if let Some(path) = args.get_one::<PathBuf>("db") {
+ smol::block_on(async { db::Database::new(path).await })?
+ } else {
+ smol::block_on(async { db::open_default_db().await })?
+ };
- if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") {
- let count = conn.get_toencode_number().await?;
- println!("Files to reencode:\t{}", style(count).green());
- return Ok(());
- }
+ let path = args.get_one::<PathBuf>("path");
- if let Some(realpath) = path {
- let newtoken = canceltoken.clone();
- files::index_files_recursively(realpath, &conn, newtoken).await?;
- }
-
- if canceltoken.is_cancelled() {
- return Ok(());
- }
+ if path.is_none() && !args.get_flag("clean") && !args.get_flag("doit") {
+ let count = smol::block_on(async { conn.get_toencode_number().await })?;
+ println!("Files to reencode:\t{}", style(count).green());
+ return Ok(());
+ }
- if args.get_flag("clean") {
- files::clean_files(&conn).await?;
- }
+ let pool = rayon::ThreadPoolBuilder::new()
+ .num_threads(*args.get_one::<usize>("threads").unwrap())
+ .build()
+ .unwrap();
- if canceltoken.is_cancelled() {
- return Ok(());
- }
+ if let Some(realpath) = path {
+ let hanlder = running.clone();
+ pool.install(|| files::index_files_recursively(realpath, &conn, hanlder))?;
+ }
- if args.get_flag("doit") {
- let newtoken = canceltoken.clone();
- files::reencode_files(&conn, newtoken).await?;
- }
- Ok::<(), anyhow::Error>(())
- })?;
+ /* if args.get_flag("clean") {
+ pool.install(|| files::clean_files(&conn))?;
+ }
- Ok(())
+ if args.get_flag("doit") {
+ files::reencode_files(&conn)?;
+ } */
+ Ok::<(), anyhow::Error>(())
}