feat!: finer control over concurrent workers (#3661)

This commit is contained in:
sxyazi 2026-02-06 22:27:08 +08:00
parent 6757fed5aa
commit dc66f79cef
No known key found for this signature in database
58 changed files with 1272 additions and 1016 deletions

View file

@ -22,6 +22,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/):
### Changed
- Upgrade Lua to 5.5 ([#3633])
- Remove `micro_workers` and `macro_workers` in favor of finer control over concurrent workers ([#3661])
### Fixed
@ -1640,3 +1641,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/):
[#3638]: https://github.com/sxyazi/yazi/pull/3638
[#3642]: https://github.com/sxyazi/yazi/pull/3642
[#3648]: https://github.com/sxyazi/yazi/pull/3648
[#3661]: https://github.com/sxyazi/yazi/pull/3661

108
Cargo.lock generated
View file

@ -159,9 +159,9 @@ dependencies = [
[[package]]
name = "anyhow"
version = "1.0.100"
version = "1.0.101"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
checksum = "5f0e0fee31ef5ed1ba1316088939cea399010ed7731dba877ed44aeb407a75ea"
[[package]]
name = "arbitrary"
@ -171,9 +171,9 @@ checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1"
[[package]]
name = "arc-swap"
version = "1.8.0"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "51d03449bb8ca2cc2ef70869af31463d1ae5ccc8fa3e334b307203fbf815207e"
checksum = "9ded5f9a03ac8f24d1b8a25101ee812cd32cdc8c50a4c50237de2c4915850e73"
dependencies = [
"rustversion",
]
@ -517,9 +517,9 @@ checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495"
[[package]]
name = "bytes"
version = "1.11.0"
version = "1.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3"
checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
[[package]]
name = "castaway"
@ -600,9 +600,9 @@ dependencies = [
[[package]]
name = "clap"
version = "4.5.56"
version = "4.5.57"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a75ca66430e33a14957acc24c5077b503e7d374151b2b4b3a10c83b4ceb4be0e"
checksum = "6899ea499e3fb9305a65d5ebf6e3d2248c5fab291f300ad0a704fbe142eae31a"
dependencies = [
"clap_builder",
"clap_derive",
@ -610,9 +610,9 @@ dependencies = [
[[package]]
name = "clap_builder"
version = "4.5.56"
version = "4.5.57"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "793207c7fa6300a0608d1080b858e5fdbe713cdc1c8db9fb17777d8a13e63df0"
checksum = "7b12c8b680195a62a8364d16b8447b01b6c2c8f9aaf68bee653be34d4245e238"
dependencies = [
"anstream",
"anstyle",
@ -920,9 +920,9 @@ dependencies = [
[[package]]
name = "crypto-common"
version = "0.2.0-rc.14"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5569e5de27d32e673283717728eb5b0e4f30898b523f151a19c51e6188006599"
checksum = "211f05e03c7d03754740fd9e585de910a095d6b99f8bcfffdef8319fa02a8331"
dependencies = [
"hybrid-array",
]
@ -1240,13 +1240,13 @@ dependencies = [
[[package]]
name = "digest"
version = "0.11.0-rc.10"
version = "0.11.0-rc.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "afa94b64bfc6549e6e4b5a3216f22593224174083da7a90db47e951c4fb31725"
checksum = "02b42f1d9edf5207c137646b568a0168ca0ec25b7f9eaf7f9961da51a3d91cea"
dependencies = [
"block-buffer 0.11.0",
"const-oid 0.10.2",
"crypto-common 0.2.0-rc.14",
"crypto-common 0.2.0",
]
[[package]]
@ -1564,9 +1564,9 @@ checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
[[package]]
name = "flate2"
version = "1.1.8"
version = "1.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b375d6465b98090a5f25b1c7703f3859783755aa9a80433b36e0379a3ec2f369"
checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
dependencies = [
"crc32fast",
"miniz_oxide",
@ -1905,9 +1905,9 @@ dependencies = [
[[package]]
name = "hybrid-array"
version = "0.4.6"
version = "0.4.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b41fb3dc24fe72c2e3a4685eed55917c2fb228851257f4a8f2d985da9443c3e5"
checksum = "e1b229d73f5803b562cc26e4da0396c8610a4ee209f4fac8fa4f8d709166dc45"
dependencies = [
"typenum",
]
@ -2064,7 +2064,7 @@ checksum = "fe44f2bbd99fcb302e246e2d6bcf51aeda346d02a365f80296a07a8c711b6da6"
dependencies = [
"argon2",
"bcrypt-pbkdf",
"digest 0.11.0-rc.10",
"digest 0.11.0-rc.11",
"ecdsa",
"ed25519-dalek",
"hex",
@ -2077,7 +2077,7 @@ dependencies = [
"rsa",
"sec1",
"sha1 0.10.6",
"sha1 0.11.0-rc.4",
"sha1 0.11.0-rc.5",
"sha2 0.10.9",
"signature 2.2.0",
"signature 3.0.0-rc.6",
@ -2990,9 +2990,9 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "pest"
version = "2.8.5"
version = "2.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2c9eb05c21a464ea704b53158d358a31e6425db2f63a1a7312268b05fe2b75f7"
checksum = "e0848c601009d37dfa3430c4666e147e49cdcf1b92ecd3e63657d8a5f19da662"
dependencies = [
"memchr",
"ucd-trie",
@ -3000,9 +3000,9 @@ dependencies = [
[[package]]
name = "pest_derive"
version = "2.8.5"
version = "2.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68f9dbced329c441fa79d80472764b1a2c7e57123553b8519b36663a2fb234ed"
checksum = "11f486f1ea21e6c10ed15d5a7c77165d0ee443402f0780849d1768e7d9d6fe77"
dependencies = [
"pest",
"pest_generator",
@ -3010,9 +3010,9 @@ dependencies = [
[[package]]
name = "pest_generator"
version = "2.8.5"
version = "2.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3bb96d5051a78f44f43c8f712d8e810adb0ebf923fc9ed2655a7f66f63ba8ee5"
checksum = "8040c4647b13b210a963c1ed407c1ff4fdfa01c31d6d2a098218702e6664f94f"
dependencies = [
"pest",
"pest_meta",
@ -3023,9 +3023,9 @@ dependencies = [
[[package]]
name = "pest_meta"
version = "2.8.5"
version = "2.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "602113b5b5e8621770cfd490cfd90b9f84ab29bd2b0e49ad83eb6d186cef2365"
checksum = "89815c69d36021a140146f26659a81d6c2afa33d216d736dd4be5381a7362220"
dependencies = [
"pest",
"sha2 0.10.9",
@ -3626,9 +3626,9 @@ dependencies = [
[[package]]
name = "regex"
version = "1.12.2"
version = "1.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "843bc0191f75f3e22651ae5f1e72939ab2f72a4bc30fa80a066bd66edefc24d4"
checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276"
dependencies = [
"aho-corasick",
"memchr",
@ -3638,9 +3638,9 @@ dependencies = [
[[package]]
name = "regex-automata"
version = "0.4.13"
version = "0.4.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5276caf25ac86c8d810222b3dbb938e512c55c6831a10f3e6ed1c93b84041f1c"
checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f"
dependencies = [
"aho-corasick",
"memchr",
@ -3649,9 +3649,9 @@ dependencies = [
[[package]]
name = "regex-syntax"
version = "0.8.8"
version = "0.8.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a2d987857b319362043e95f5353c0535c1f58eec5336fdfcf626430af7def58"
checksum = "a96887878f22d7bad8a3b6dc5b7440e0ada9a245242924394987b21cf2210a4c"
[[package]]
name = "rfc6979"
@ -3692,11 +3692,11 @@ dependencies = [
"const-oid 0.10.2",
"crypto-bigint 0.7.0-rc.18",
"crypto-primes",
"digest 0.11.0-rc.10",
"digest 0.11.0-rc.11",
"pkcs1",
"pkcs8 0.11.0-rc.10",
"rand_core 0.10.0-rc-3",
"sha2 0.11.0-rc.4",
"sha2 0.11.0-rc.5",
"signature 3.0.0-rc.6",
"spki 0.8.0-rc.4",
"zeroize",
@ -4051,13 +4051,13 @@ dependencies = [
[[package]]
name = "sha1"
version = "0.11.0-rc.4"
version = "0.11.0-rc.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c777f0a122a53fddb0beb6e706771197000b8eb5c9f42b5b850f450ef48c788"
checksum = "3b167252f3c126be0d8926639c4c4706950f01445900c4b3db0fd7e89fcb750a"
dependencies = [
"cfg-if",
"cpufeatures",
"digest 0.11.0-rc.10",
"digest 0.11.0-rc.11",
]
[[package]]
@ -4073,13 +4073,13 @@ dependencies = [
[[package]]
name = "sha2"
version = "0.11.0-rc.4"
version = "0.11.0-rc.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7535f94fa3339fe9e5e9be6260a909e62af97f6e14b32345ccf79b92b8b81233"
checksum = "7c5f3b1e2dc8aad28310d8410bd4d7e180eca65fca176c52ab00d364475d0024"
dependencies = [
"cfg-if",
"cpufeatures",
"digest 0.11.0-rc.10",
"digest 0.11.0-rc.11",
]
[[package]]
@ -4166,7 +4166,7 @@ version = "3.0.0-rc.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "597a96996ccff7dfa16f052bd995b4cecc72af22c35138738dc029f0ead6608d"
dependencies = [
"digest 0.11.0-rc.10",
"digest 0.11.0-rc.11",
"rand_core 0.10.0-rc-3",
]
@ -4527,9 +4527,9 @@ dependencies = [
[[package]]
name = "time"
version = "0.3.46"
version = "0.3.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9da98b7d9b7dad93488a84b8248efc35352b0b2657397d4167e7ad67e5d535e5"
checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c"
dependencies = [
"deranged",
"itoa",
@ -4550,9 +4550,9 @@ checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca"
[[package]]
name = "time-macros"
version = "0.2.26"
version = "0.2.27"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78cc610bac2dcee56805c99642447d4c5dbde4d01f752ffea0199aee1f601dc4"
checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215"
dependencies = [
"num-conv",
"time-core",
@ -6120,18 +6120,18 @@ dependencies = [
[[package]]
name = "zerocopy"
version = "0.8.37"
version = "0.8.39"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7456cf00f0685ad319c5b1693f291a650eaf345e941d082fc4e03df8a03996ac"
checksum = "db6d35d663eadb6c932438e763b262fe1a70987f9ae936e60158176d710cae4a"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
version = "0.8.37"
version = "0.8.39"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1328722bbf2115db7e19d69ebcc15e795719e2d66b60827c6a69a117365e37a0"
checksum = "4122cd3169e94605190e77839c9a40d40ed048d305bfdc146e7df40ab0f3e517"
dependencies = [
"proc-macro2",
"quote",
@ -6160,9 +6160,9 @@ dependencies = [
[[package]]
name = "zmij"
version = "1.0.18"
version = "1.0.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1966f8ac2c1f76987d69a74d0e0f929241c10e78136434e3be70ff7f58f64214"
checksum = "3ff05f8caa9038894637571ae6b9e29466c1f4f829d26c9b28f869a29cbe3445"
[[package]]
name = "zune-core"

View file

@ -27,10 +27,10 @@ debug = false
[workspace.dependencies]
ansi-to-tui = "8.0.1"
anyhow = "1.0.100"
anyhow = "1.0.101"
base64 = "0.22.1"
bitflags = { version = "2.10.0", features = [ "serde" ] }
clap = { version = "4.5.56", features = [ "derive" ] }
clap = { version = "4.5.57", features = [ "derive" ] }
core-foundation-sys = "0.8.7"
crossterm = { version = "0.29.0", features = [ "event-stream" ] }
dirs = "6.0.0"
@ -51,7 +51,7 @@ paste = "1.0.15"
percent-encoding = "2.3.2"
rand = { version = "0.9.2", default-features = false, features = [ "os_rng", "small_rng", "std" ] }
ratatui = { version = "0.30.0", features = [ "serde", "unstable-rendered-line-info", "unstable-widget-ref" ] }
regex = "1.12.2"
regex = "1.12.3"
russh = { version = "0.57.0", default-features = false, features = [ "ring", "rsa" ] }
scopeguard = "1.2.0"
serde = { version = "1.0.228", features = [ "derive" ] }

View file

@ -8,247 +8,218 @@
"name": "validate-form",
"version": "1.0.0",
"dependencies": {
"@actions/core": "^2.0.2",
"@actions/github": "^7.0.0"
"@actions/core": "^3.0.0",
"@actions/github": "^9.0.0"
}
},
"node_modules/@actions/core": {
"version": "2.0.2",
"resolved": "https://registry.npmjs.org/@actions/core/-/core-2.0.2.tgz",
"integrity": "sha512-Ast1V7yHbGAhplAsuVlnb/5J8Mtr/Zl6byPPL+Qjq3lmfIgWF1ak1iYfF/079cRERiuTALTXkSuEUdZeDCfGtA==",
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/@actions/core/-/core-3.0.0.tgz",
"integrity": "sha512-zYt6cz+ivnTmiT/ksRVriMBOiuoUpDCJJlZ5KPl2/FRdvwU3f7MPh9qftvbkXJThragzUZieit2nyHUyw53Seg==",
"license": "MIT",
"dependencies": {
"@actions/exec": "^2.0.0",
"@actions/http-client": "^3.0.1"
"@actions/exec": "^3.0.0",
"@actions/http-client": "^4.0.0"
}
},
"node_modules/@actions/exec": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/@actions/exec/-/exec-2.0.0.tgz",
"integrity": "sha512-k8ngrX2voJ/RIN6r9xB82NVqKpnMRtxDoiO+g3olkIUpQNqjArXrCQceduQZCQj3P3xm32pChRLqRrtXTlqhIw==",
"license": "MIT",
"dependencies": {
"@actions/io": "^2.0.0"
}
},
"node_modules/@actions/github": {
"version": "7.0.0",
"resolved": "https://registry.npmjs.org/@actions/github/-/github-7.0.0.tgz",
"integrity": "sha512-PyGODO938aoBTZd/IfN/+e+Pd5hUcVpyf+thm4CPESLeqhdSkq5QwMTGX9v84XHE1ifmHWBQ60KB8kIgm96opw==",
"license": "MIT",
"dependencies": {
"@actions/http-client": "^3.0.1",
"@octokit/core": "^5.0.1",
"@octokit/plugin-paginate-rest": "^9.2.2",
"@octokit/plugin-rest-endpoint-methods": "^10.4.0",
"@octokit/request": "^8.4.1",
"@octokit/request-error": "^5.1.1",
"undici": "^5.28.5"
}
},
"node_modules/@actions/http-client": {
"version": "3.0.1",
"resolved": "https://registry.npmjs.org/@actions/http-client/-/http-client-3.0.1.tgz",
"integrity": "sha512-SbGS8c/vySbNO3kjFgSW77n83C4MQx/Yoe+b1hAdpuvfHxnkHzDq2pWljUpAA56Si1Gae/7zjeZsV0CYjmLo/w==",
"node_modules/@actions/core/node_modules/@actions/http-client": {
"version": "4.0.0",
"resolved": "https://registry.npmjs.org/@actions/http-client/-/http-client-4.0.0.tgz",
"integrity": "sha512-QuwPsgVMsD6qaPD57GLZi9sqzAZCtiJT8kVBCDpLtxhL5MydQ4gS+DrejtZZPdIYyB1e95uCK9Luyds7ybHI3g==",
"license": "MIT",
"dependencies": {
"tunnel": "^0.0.6",
"undici": "^5.28.5"
"undici": "^6.23.0"
}
},
"node_modules/@actions/exec": {
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/@actions/exec/-/exec-3.0.0.tgz",
"integrity": "sha512-6xH/puSoNBXb72VPlZVm7vQ+svQpFyA96qdDBvhB8eNZOE8LtPf9L4oAsfzK/crCL8YZ+19fKYVnM63Sl+Xzlw==",
"license": "MIT",
"dependencies": {
"@actions/io": "^3.0.2"
}
},
"node_modules/@actions/github": {
"version": "9.0.0",
"resolved": "https://registry.npmjs.org/@actions/github/-/github-9.0.0.tgz",
"integrity": "sha512-yJ0RoswsAaKcvkmpCE4XxBRiy/whH2SdTBHWzs0gi4wkqTDhXMChjSdqBz/F4AeiDlP28rQqL33iHb+kjAMX6w==",
"license": "MIT",
"dependencies": {
"@actions/http-client": "^3.0.2",
"@octokit/core": "^7.0.6",
"@octokit/plugin-paginate-rest": "^14.0.0",
"@octokit/plugin-rest-endpoint-methods": "^17.0.0",
"@octokit/request": "^10.0.7",
"@octokit/request-error": "^7.1.0",
"undici": "^6.23.0"
}
},
"node_modules/@actions/http-client": {
"version": "3.0.2",
"resolved": "https://registry.npmjs.org/@actions/http-client/-/http-client-3.0.2.tgz",
"integrity": "sha512-JP38FYYpyqvUsz+Igqlc/JG6YO9PaKuvqjM3iGvaLqFnJ7TFmcLyy2IDrY0bI0qCQug8E9K+elv5ZNfw62ZJzA==",
"license": "MIT",
"dependencies": {
"tunnel": "^0.0.6",
"undici": "^6.23.0"
}
},
"node_modules/@actions/io": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/@actions/io/-/io-2.0.0.tgz",
"integrity": "sha512-Jv33IN09XLO+0HS79aaODsvIRyduiF7NY/F6LYeK5oeUmrsz7aFdRphQjFoESF4jS7lMauDOttKALcpapVDIAg==",
"version": "3.0.2",
"resolved": "https://registry.npmjs.org/@actions/io/-/io-3.0.2.tgz",
"integrity": "sha512-nRBchcMM+QK1pdjO7/idu86rbJI5YHUKCvKs0KxnSYbVe3F51UfGxuZX4Qy/fWlp6l7gWFwIkrOzN+oUK03kfw==",
"license": "MIT"
},
"node_modules/@fastify/busboy": {
"version": "2.1.1",
"resolved": "https://registry.npmjs.org/@fastify/busboy/-/busboy-2.1.1.tgz",
"integrity": "sha512-vBZP4NlzfOlerQTnba4aqZoMhE/a9HY7HRqoOPaETQcSQuWEIyZMHGfVu6w9wGtGK5fED5qRs2DteVCjOH60sA==",
"license": "MIT",
"engines": {
"node": ">=14"
}
},
"node_modules/@octokit/auth-token": {
"version": "4.0.0",
"resolved": "https://registry.npmjs.org/@octokit/auth-token/-/auth-token-4.0.0.tgz",
"integrity": "sha512-tY/msAuJo6ARbK6SPIxZrPBms3xPbfwBrulZe0Wtr/DIY9lje2HeV1uoebShn6mx7SjCHif6EjMvoREj+gZ+SA==",
"version": "6.0.0",
"resolved": "https://registry.npmjs.org/@octokit/auth-token/-/auth-token-6.0.0.tgz",
"integrity": "sha512-P4YJBPdPSpWTQ1NU4XYdvHvXJJDxM6YwpS0FZHRgP7YFkdVxsWcpWGy/NVqlAA7PcPCnMacXlRm1y2PFZRWL/w==",
"license": "MIT",
"engines": {
"node": ">= 18"
"node": ">= 20"
}
},
"node_modules/@octokit/core": {
"version": "5.2.2",
"resolved": "https://registry.npmjs.org/@octokit/core/-/core-5.2.2.tgz",
"integrity": "sha512-/g2d4sW9nUDJOMz3mabVQvOGhVa4e/BN/Um7yca9Bb2XTzPPnfTWHWQg+IsEYO7M3Vx+EXvaM/I2pJWIMun1bg==",
"version": "7.0.6",
"resolved": "https://registry.npmjs.org/@octokit/core/-/core-7.0.6.tgz",
"integrity": "sha512-DhGl4xMVFGVIyMwswXeyzdL4uXD5OGILGX5N8Y+f6W7LhC1Ze2poSNrkF/fedpVDHEEZ+PHFW0vL14I+mm8K3Q==",
"license": "MIT",
"dependencies": {
"@octokit/auth-token": "^4.0.0",
"@octokit/graphql": "^7.1.0",
"@octokit/request": "^8.4.1",
"@octokit/request-error": "^5.1.1",
"@octokit/types": "^13.0.0",
"before-after-hook": "^2.2.0",
"universal-user-agent": "^6.0.0"
"@octokit/auth-token": "^6.0.0",
"@octokit/graphql": "^9.0.3",
"@octokit/request": "^10.0.6",
"@octokit/request-error": "^7.0.2",
"@octokit/types": "^16.0.0",
"before-after-hook": "^4.0.0",
"universal-user-agent": "^7.0.0"
},
"engines": {
"node": ">= 18"
"node": ">= 20"
}
},
"node_modules/@octokit/endpoint": {
"version": "9.0.6",
"resolved": "https://registry.npmjs.org/@octokit/endpoint/-/endpoint-9.0.6.tgz",
"integrity": "sha512-H1fNTMA57HbkFESSt3Y9+FBICv+0jFceJFPWDePYlR/iMGrwM5ph+Dd4XRQs+8X+PUFURLQgX9ChPfhJ/1uNQw==",
"version": "11.0.2",
"resolved": "https://registry.npmjs.org/@octokit/endpoint/-/endpoint-11.0.2.tgz",
"integrity": "sha512-4zCpzP1fWc7QlqunZ5bSEjxc6yLAlRTnDwKtgXfcI/FxxGoqedDG8V2+xJ60bV2kODqcGB+nATdtap/XYq2NZQ==",
"license": "MIT",
"dependencies": {
"@octokit/types": "^13.1.0",
"universal-user-agent": "^6.0.0"
"@octokit/types": "^16.0.0",
"universal-user-agent": "^7.0.2"
},
"engines": {
"node": ">= 18"
"node": ">= 20"
}
},
"node_modules/@octokit/graphql": {
"version": "7.1.1",
"resolved": "https://registry.npmjs.org/@octokit/graphql/-/graphql-7.1.1.tgz",
"integrity": "sha512-3mkDltSfcDUoa176nlGoA32RGjeWjl3K7F/BwHwRMJUW/IteSa4bnSV8p2ThNkcIcZU2umkZWxwETSSCJf2Q7g==",
"version": "9.0.3",
"resolved": "https://registry.npmjs.org/@octokit/graphql/-/graphql-9.0.3.tgz",
"integrity": "sha512-grAEuupr/C1rALFnXTv6ZQhFuL1D8G5y8CN04RgrO4FIPMrtm+mcZzFG7dcBm+nq+1ppNixu+Jd78aeJOYxlGA==",
"license": "MIT",
"dependencies": {
"@octokit/request": "^8.4.1",
"@octokit/types": "^13.0.0",
"universal-user-agent": "^6.0.0"
"@octokit/request": "^10.0.6",
"@octokit/types": "^16.0.0",
"universal-user-agent": "^7.0.0"
},
"engines": {
"node": ">= 18"
"node": ">= 20"
}
},
"node_modules/@octokit/openapi-types": {
"version": "24.2.0",
"resolved": "https://registry.npmjs.org/@octokit/openapi-types/-/openapi-types-24.2.0.tgz",
"integrity": "sha512-9sIH3nSUttelJSXUrmGzl7QUBFul0/mB8HRYl3fOlgHbIWG+WnYDXU3v/2zMtAvuzZ/ed00Ei6on975FhBfzrg==",
"version": "27.0.0",
"resolved": "https://registry.npmjs.org/@octokit/openapi-types/-/openapi-types-27.0.0.tgz",
"integrity": "sha512-whrdktVs1h6gtR+09+QsNk2+FO+49j6ga1c55YZudfEG+oKJVvJLQi3zkOm5JjiUXAagWK2tI2kTGKJ2Ys7MGA==",
"license": "MIT"
},
"node_modules/@octokit/plugin-paginate-rest": {
"version": "9.2.2",
"resolved": "https://registry.npmjs.org/@octokit/plugin-paginate-rest/-/plugin-paginate-rest-9.2.2.tgz",
"integrity": "sha512-u3KYkGF7GcZnSD/3UP0S7K5XUFT2FkOQdcfXZGZQPGv3lm4F2Xbf71lvjldr8c1H3nNbF+33cLEkWYbokGWqiQ==",
"version": "14.0.0",
"resolved": "https://registry.npmjs.org/@octokit/plugin-paginate-rest/-/plugin-paginate-rest-14.0.0.tgz",
"integrity": "sha512-fNVRE7ufJiAA3XUrha2omTA39M6IXIc6GIZLvlbsm8QOQCYvpq/LkMNGyFlB1d8hTDzsAXa3OKtybdMAYsV/fw==",
"license": "MIT",
"dependencies": {
"@octokit/types": "^12.6.0"
"@octokit/types": "^16.0.0"
},
"engines": {
"node": ">= 18"
"node": ">= 20"
},
"peerDependencies": {
"@octokit/core": "5"
}
},
"node_modules/@octokit/plugin-paginate-rest/node_modules/@octokit/openapi-types": {
"version": "20.0.0",
"resolved": "https://registry.npmjs.org/@octokit/openapi-types/-/openapi-types-20.0.0.tgz",
"integrity": "sha512-EtqRBEjp1dL/15V7WiX5LJMIxxkdiGJnabzYx5Apx4FkQIFgAfKumXeYAqqJCj1s+BMX4cPFIFC4OLCR6stlnA==",
"license": "MIT"
},
"node_modules/@octokit/plugin-paginate-rest/node_modules/@octokit/types": {
"version": "12.6.0",
"resolved": "https://registry.npmjs.org/@octokit/types/-/types-12.6.0.tgz",
"integrity": "sha512-1rhSOfRa6H9w4YwK0yrf5faDaDTb+yLyBUKOCV4xtCDB5VmIPqd/v9yr9o6SAzOAlRxMiRiCic6JVM1/kunVkw==",
"license": "MIT",
"dependencies": {
"@octokit/openapi-types": "^20.0.0"
"@octokit/core": ">=6"
}
},
"node_modules/@octokit/plugin-rest-endpoint-methods": {
"version": "10.4.1",
"resolved": "https://registry.npmjs.org/@octokit/plugin-rest-endpoint-methods/-/plugin-rest-endpoint-methods-10.4.1.tgz",
"integrity": "sha512-xV1b+ceKV9KytQe3zCVqjg+8GTGfDYwaT1ATU5isiUyVtlVAO3HNdzpS4sr4GBx4hxQ46s7ITtZrAsxG22+rVg==",
"version": "17.0.0",
"resolved": "https://registry.npmjs.org/@octokit/plugin-rest-endpoint-methods/-/plugin-rest-endpoint-methods-17.0.0.tgz",
"integrity": "sha512-B5yCyIlOJFPqUUeiD0cnBJwWJO8lkJs5d8+ze9QDP6SvfiXSz1BF+91+0MeI1d2yxgOhU/O+CvtiZ9jSkHhFAw==",
"license": "MIT",
"dependencies": {
"@octokit/types": "^12.6.0"
"@octokit/types": "^16.0.0"
},
"engines": {
"node": ">= 18"
"node": ">= 20"
},
"peerDependencies": {
"@octokit/core": "5"
}
},
"node_modules/@octokit/plugin-rest-endpoint-methods/node_modules/@octokit/openapi-types": {
"version": "20.0.0",
"resolved": "https://registry.npmjs.org/@octokit/openapi-types/-/openapi-types-20.0.0.tgz",
"integrity": "sha512-EtqRBEjp1dL/15V7WiX5LJMIxxkdiGJnabzYx5Apx4FkQIFgAfKumXeYAqqJCj1s+BMX4cPFIFC4OLCR6stlnA==",
"license": "MIT"
},
"node_modules/@octokit/plugin-rest-endpoint-methods/node_modules/@octokit/types": {
"version": "12.6.0",
"resolved": "https://registry.npmjs.org/@octokit/types/-/types-12.6.0.tgz",
"integrity": "sha512-1rhSOfRa6H9w4YwK0yrf5faDaDTb+yLyBUKOCV4xtCDB5VmIPqd/v9yr9o6SAzOAlRxMiRiCic6JVM1/kunVkw==",
"license": "MIT",
"dependencies": {
"@octokit/openapi-types": "^20.0.0"
"@octokit/core": ">=6"
}
},
"node_modules/@octokit/request": {
"version": "8.4.1",
"resolved": "https://registry.npmjs.org/@octokit/request/-/request-8.4.1.tgz",
"integrity": "sha512-qnB2+SY3hkCmBxZsR/MPCybNmbJe4KAlfWErXq+rBKkQJlbjdJeS85VI9r8UqeLYLvnAenU8Q1okM/0MBsAGXw==",
"version": "10.0.7",
"resolved": "https://registry.npmjs.org/@octokit/request/-/request-10.0.7.tgz",
"integrity": "sha512-v93h0i1yu4idj8qFPZwjehoJx4j3Ntn+JhXsdJrG9pYaX6j/XRz2RmasMUHtNgQD39nrv/VwTWSqK0RNXR8upA==",
"license": "MIT",
"dependencies": {
"@octokit/endpoint": "^9.0.6",
"@octokit/request-error": "^5.1.1",
"@octokit/types": "^13.1.0",
"universal-user-agent": "^6.0.0"
"@octokit/endpoint": "^11.0.2",
"@octokit/request-error": "^7.0.2",
"@octokit/types": "^16.0.0",
"fast-content-type-parse": "^3.0.0",
"universal-user-agent": "^7.0.2"
},
"engines": {
"node": ">= 18"
"node": ">= 20"
}
},
"node_modules/@octokit/request-error": {
"version": "5.1.1",
"resolved": "https://registry.npmjs.org/@octokit/request-error/-/request-error-5.1.1.tgz",
"integrity": "sha512-v9iyEQJH6ZntoENr9/yXxjuezh4My67CBSu9r6Ve/05Iu5gNgnisNWOsoJHTP6k0Rr0+HQIpnH+kyammu90q/g==",
"version": "7.1.0",
"resolved": "https://registry.npmjs.org/@octokit/request-error/-/request-error-7.1.0.tgz",
"integrity": "sha512-KMQIfq5sOPpkQYajXHwnhjCC0slzCNScLHs9JafXc4RAJI+9f+jNDlBNaIMTvazOPLgb4BnlhGJOTbnN0wIjPw==",
"license": "MIT",
"dependencies": {
"@octokit/types": "^13.1.0",
"deprecation": "^2.0.0",
"once": "^1.4.0"
"@octokit/types": "^16.0.0"
},
"engines": {
"node": ">= 18"
"node": ">= 20"
}
},
"node_modules/@octokit/types": {
"version": "13.10.0",
"resolved": "https://registry.npmjs.org/@octokit/types/-/types-13.10.0.tgz",
"integrity": "sha512-ifLaO34EbbPj0Xgro4G5lP5asESjwHracYJvVaPIyXMuiuXLlhic3S47cBdTb+jfODkTE5YtGCLt3Ay3+J97sA==",
"version": "16.0.0",
"resolved": "https://registry.npmjs.org/@octokit/types/-/types-16.0.0.tgz",
"integrity": "sha512-sKq+9r1Mm4efXW1FCk7hFSeJo4QKreL/tTbR0rz/qx/r1Oa2VV83LTA/H/MuCOX7uCIJmQVRKBcbmWoySjAnSg==",
"license": "MIT",
"dependencies": {
"@octokit/openapi-types": "^24.2.0"
"@octokit/openapi-types": "^27.0.0"
}
},
"node_modules/before-after-hook": {
"version": "2.2.3",
"resolved": "https://registry.npmjs.org/before-after-hook/-/before-after-hook-2.2.3.tgz",
"integrity": "sha512-NzUnlZexiaH/46WDhANlyR2bXRopNg4F/zuSA3OpZnllCUgRaOF2znDioDWrmbNVsuZk6l9pMquQB38cfBZwkQ==",
"version": "4.0.0",
"resolved": "https://registry.npmjs.org/before-after-hook/-/before-after-hook-4.0.0.tgz",
"integrity": "sha512-q6tR3RPqIB1pMiTRMFcZwuG5T8vwp+vUvEG0vuI6B+Rikh5BfPp2fQ82c925FOs+b0lcFQ8CFrL+KbilfZFhOQ==",
"license": "Apache-2.0"
},
"node_modules/deprecation": {
"version": "2.3.1",
"resolved": "https://registry.npmjs.org/deprecation/-/deprecation-2.3.1.tgz",
"integrity": "sha512-xmHIy4F3scKVwMsQ4WnVaS8bHOx0DmVwRywosKhaILI0ywMDWPtBSku2HNxRvF7jtwDRsoEwYQSfbxj8b7RlJQ==",
"license": "ISC"
},
"node_modules/once": {
"version": "1.4.0",
"resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz",
"integrity": "sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==",
"license": "ISC",
"dependencies": {
"wrappy": "1"
}
"node_modules/fast-content-type-parse": {
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/fast-content-type-parse/-/fast-content-type-parse-3.0.0.tgz",
"integrity": "sha512-ZvLdcY8P+N8mGQJahJV5G4U88CSvT1rP8ApL6uETe88MBXrBHAkZlSEySdUlyztF7ccb+Znos3TFqaepHxdhBg==",
"funding": [
{
"type": "github",
"url": "https://github.com/sponsors/fastify"
},
{
"type": "opencollective",
"url": "https://opencollective.com/fastify"
}
],
"license": "MIT"
},
"node_modules/tunnel": {
"version": "0.0.6",
@ -260,27 +231,18 @@
}
},
"node_modules/undici": {
"version": "5.29.0",
"resolved": "https://registry.npmjs.org/undici/-/undici-5.29.0.tgz",
"integrity": "sha512-raqeBD6NQK4SkWhQzeYKd1KmIG6dllBOTt55Rmkt4HtI9mwdWtJljnrXjAFUBLTSN67HWrOIZ3EPF4kjUw80Bg==",
"version": "6.23.0",
"resolved": "https://registry.npmjs.org/undici/-/undici-6.23.0.tgz",
"integrity": "sha512-VfQPToRA5FZs/qJxLIinmU59u0r7LXqoJkCzinq3ckNJp3vKEh7jTWN589YQ5+aoAC/TGRLyJLCPKcLQbM8r9g==",
"license": "MIT",
"dependencies": {
"@fastify/busboy": "^2.0.0"
},
"engines": {
"node": ">=14.0"
"node": ">=18.17"
}
},
"node_modules/universal-user-agent": {
"version": "6.0.1",
"resolved": "https://registry.npmjs.org/universal-user-agent/-/universal-user-agent-6.0.1.tgz",
"integrity": "sha512-yCzhz6FN2wU1NiiQRogkTQszlQSlpWaw8SvVegAc+bDxbzHgh1vX8uIe8OYyMH6DwH+sdTJsgMl36+mSMdRJIQ==",
"license": "ISC"
},
"node_modules/wrappy": {
"version": "1.0.2",
"resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz",
"integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==",
"version": "7.0.3",
"resolved": "https://registry.npmjs.org/universal-user-agent/-/universal-user-agent-7.0.3.tgz",
"integrity": "sha512-TmnEAEAsBJVZM/AADELsK76llnwcf9vMKuPz8JflO1frO8Lchitr0fNaN9d+Ap0BjKtqWqd/J17qeDnXh8CL2A==",
"license": "ISC"
}
}

View file

@ -6,7 +6,7 @@
},
"private": true,
"dependencies": {
"@actions/core": "^2.0.2",
"@actions/github": "^7.0.0"
"@actions/core": "^3.0.0",
"@actions/github": "^9.0.0"
}
}

View file

@ -87,8 +87,11 @@ rules = [
]
[tasks]
micro_workers = 10
macro_workers = 10
file_workers = 3
plugin_workers = 5
fetch_workers = 5
preload_workers = 2
process_workers = 5
bizarre_retry = 3
image_alloc = 536870912 # 512MB
image_bound = [ 10000, 10000 ]

View file

@ -1,3 +1,4 @@
yazi_macro::mod_flat!(fetcher plugin preloader previewer spotter);
pub const MAX_PREWORKERS: u8 = 32;
pub const MAX_FETCHERS: u8 = 16;
pub const MAX_PRELOADERS: u8 = 16;

View file

@ -1,4 +1,4 @@
use anyhow::Result;
use anyhow::{Result, bail};
use hashbrown::HashSet;
use serde::Deserialize;
use tracing::warn;
@ -6,7 +6,7 @@ use yazi_codegen::DeserializeOver2;
use yazi_fs::File;
use super::{Fetcher, Preloader, Previewer, Spotter};
use crate::{Preset, plugin::MAX_PREWORKERS};
use crate::{Preset, plugin::{MAX_FETCHERS, MAX_PRELOADERS}};
#[derive(Default, Deserialize, DeserializeOver2)]
pub struct Plugin {
@ -52,7 +52,7 @@ impl Plugin {
}
pub fn mime_fetchers(&self, files: Vec<File>) -> impl Iterator<Item = (&Fetcher, Vec<File>)> {
let mut tasks: [Vec<_>; MAX_PREWORKERS as usize] = Default::default();
let mut tasks: [Vec<_>; MAX_FETCHERS as usize] = Default::default();
for f in files {
let found = self.fetchers.iter().find(|&g| g.id == "mime" && g.matches(&f, ""));
if let Some(g) = found {
@ -116,15 +116,17 @@ impl Plugin {
self.previewers =
Preset::mix(self.prepend_previewers, self.previewers, self.append_previewers).collect();
if self.fetchers.len() + self.preloaders.len() > MAX_PREWORKERS as usize {
panic!("Fetchers and preloaders exceed the limit of {MAX_PREWORKERS}");
if self.fetchers.len() > MAX_FETCHERS as usize {
bail!("Fetchers exceed the limit of {MAX_FETCHERS}");
} else if self.preloaders.len() > MAX_PRELOADERS as usize {
bail!("Preloaders exceed the limit of {MAX_PRELOADERS}");
}
for (i, p) in self.fetchers.iter_mut().enumerate() {
p.idx = i as u8;
}
for (i, p) in self.preloaders.iter_mut().enumerate() {
p.idx = self.fetchers.len() as u8 + i as u8;
p.idx = i as u8;
}
Ok(Self {

View file

@ -4,8 +4,12 @@ use yazi_codegen::DeserializeOver2;
#[derive(Debug, Deserialize, DeserializeOver2)]
pub struct Tasks {
pub micro_workers: u8,
pub macro_workers: u8,
pub file_workers: u8,
pub plugin_workers: u8,
pub fetch_workers: u8,
pub preload_workers: u8,
pub process_workers: u8,
pub bizarre_retry: u8,
pub image_alloc: u32,
@ -16,12 +20,18 @@ pub struct Tasks {
impl Tasks {
pub(crate) fn reshape(self) -> Result<Self> {
if self.micro_workers < 1 {
bail!("[tasks].micro_workers must be at least 1.");
} else if self.macro_workers < 1 {
bail!("[tasks].macro_workers must be at least 1.");
if self.file_workers < 1 {
bail!("[tasks].file_workers must be at least 1.");
} else if self.plugin_workers < 1 {
bail!("[tasks].plugin_workers must be at least 1.");
} else if self.fetch_workers < 1 {
bail!("[tasks].fetch_workers must be at least 1.");
} else if self.preload_workers < 1 {
bail!("[tasks].preload_workers must be at least 1.");
} else if self.process_workers < 1 {
bail!("[tasks].process_workers must be at least 1.");
} else if self.bizarre_retry < 1 {
bail!("[tasks].bizarre_retry` must be at least 1.");
bail!("[tasks].bizarre_retry must be at least 1.");
}
Ok(self)
}

View file

@ -1,4 +1,4 @@
use yazi_config::{YAZI, plugin::MAX_PREWORKERS};
use yazi_config::{YAZI, plugin::MAX_FETCHERS};
use yazi_fs::{File, Files, FsHash64, SortBy};
use super::Tasks;
@ -6,8 +6,8 @@ use crate::mgr::Mimetype;
impl Tasks {
pub fn fetch_paged(&self, paged: &[File], mimetype: &Mimetype) {
let mut loaded = self.scheduler.runner.prework.loaded.lock();
let mut tasks: [Vec<_>; MAX_PREWORKERS as usize] = Default::default();
let mut loaded = self.scheduler.fetch.loaded.lock();
let mut tasks: [Vec<_>; MAX_FETCHERS as usize] = Default::default();
for f in paged {
let hash = f.hash_u64();
for g in YAZI.plugin.fetchers(f, mimetype.get(&f.url).unwrap_or_default()) {
@ -29,7 +29,7 @@ impl Tasks {
}
pub fn preload_paged(&self, paged: &[File], mimetype: &Mimetype) {
let mut loaded = self.scheduler.runner.prework.loaded.lock();
let mut loaded = self.scheduler.preload.loaded.lock();
for f in paged {
let hash = f.hash_u64();
for p in YAZI.plugin.preloaders(f, mimetype.get(&f.url).unwrap_or_default()) {
@ -49,7 +49,7 @@ impl Tasks {
}
let targets: Vec<_> = {
let loading = self.scheduler.runner.prework.sizing.read();
let loading = self.scheduler.size.sizing.read();
targets
.iter()
.filter(|f| {
@ -62,7 +62,7 @@ impl Tasks {
return;
}
let mut loading = self.scheduler.runner.prework.sizing.write();
let mut loading = self.scheduler.size.sizing.write();
for &target in &targets {
loading.insert(target.clone());
}

View file

@ -19,7 +19,7 @@ yazi-shim = { path = "../yazi-shim", version = "26.2.2" }
# External dependencies
anyhow = { workspace = true }
arc-swap = "1.8.0"
arc-swap = "1.8.1"
bitflags = { workspace = true }
dirs = { workspace = true }
either = { workspace = true }

View file

@ -44,17 +44,17 @@ fn mgr() -> Composer<ComposerGet, ComposerSet> {
b"ratio" => lua.to_value_with(&m.ratio, SER_OPT)?,
b"sort_by" => lua.to_value_with(&m.sort_by, SER_OPT)?,
b"sort_sensitive" => lua.to_value_with(&m.sort_sensitive, SER_OPT)?,
b"sort_reverse" => lua.to_value_with(&m.sort_reverse, SER_OPT)?,
b"sort_dir_first" => lua.to_value_with(&m.sort_dir_first, SER_OPT)?,
b"sort_translit" => lua.to_value_with(&m.sort_translit, SER_OPT)?,
b"sort_sensitive" => m.sort_sensitive.get().into_lua(lua)?,
b"sort_reverse" => m.sort_reverse.get().into_lua(lua)?,
b"sort_dir_first" => m.sort_dir_first.get().into_lua(lua)?,
b"sort_translit" => m.sort_translit.get().into_lua(lua)?,
b"linemode" => lua.to_value_with(&m.linemode, SER_OPT)?,
b"show_hidden" => lua.to_value_with(&m.show_hidden, SER_OPT)?,
b"show_symlink" => lua.to_value_with(&m.show_symlink, SER_OPT)?,
b"scrolloff" => lua.to_value_with(&m.scrolloff, SER_OPT)?,
b"linemode" => lua.create_string(&m.linemode)?.into_lua(lua)?,
b"show_hidden" => m.show_hidden.get().into_lua(lua)?,
b"show_symlink" => m.show_symlink.get().into_lua(lua)?,
b"scrolloff" => m.scrolloff.get().into_lua(lua)?,
b"mouse_events" => lua.to_value_with(&m.mouse_events, SER_OPT)?,
b"title_format" => lua.to_value_with(&m.title_format, SER_OPT)?,
b"title_format" => lua.create_string(&m.title_format)?.into_lua(lua)?,
_ => return Ok(Value::Nil),
}
.into_lua(lua)
@ -79,17 +79,17 @@ fn preview() -> Composer<ComposerGet, ComposerSet> {
let p = &YAZI.preview;
match key {
b"wrap" => lua.to_value_with(&p.wrap, SER_OPT)?,
b"tab_size" => lua.to_value_with(&p.tab_size, SER_OPT)?,
b"max_width" => lua.to_value_with(&p.max_width, SER_OPT)?,
b"max_height" => lua.to_value_with(&p.max_height, SER_OPT)?,
b"tab_size" => p.tab_size.into_lua(lua)?,
b"max_width" => p.max_width.into_lua(lua)?,
b"max_height" => p.max_height.into_lua(lua)?,
b"cache_dir" => lua.to_value_with(&p.cache_dir, SER_OPT)?,
b"image_delay" => lua.to_value_with(&p.image_delay, SER_OPT)?,
b"image_filter" => lua.to_value_with(&p.image_filter, SER_OPT)?,
b"image_quality" => lua.to_value_with(&p.image_quality, SER_OPT)?,
b"image_delay" => p.image_delay.into_lua(lua)?,
b"image_filter" => lua.create_string(&p.image_filter)?.into_lua(lua)?,
b"image_quality" => p.image_quality.into_lua(lua)?,
b"ueberzug_scale" => lua.to_value_with(&p.ueberzug_scale, SER_OPT)?,
b"ueberzug_scale" => p.ueberzug_scale.into_lua(lua)?,
b"ueberzug_offset" => lua.to_value_with(&p.ueberzug_offset, SER_OPT)?,
_ => return Ok(Value::Nil),
}
@ -105,14 +105,17 @@ fn tasks() -> Composer<ComposerGet, ComposerSet> {
fn get(lua: &Lua, key: &[u8]) -> mlua::Result<Value> {
let t = &YAZI.tasks;
match key {
b"micro_workers" => lua.to_value_with(&t.micro_workers, SER_OPT)?,
b"macro_workers" => lua.to_value_with(&t.macro_workers, SER_OPT)?,
b"bizarre_retry" => lua.to_value_with(&t.bizarre_retry, SER_OPT)?,
b"file_workers" => t.file_workers.into_lua(lua)?,
b"plugin_workers" => t.plugin_workers.into_lua(lua)?,
b"fetch_workers" => t.fetch_workers.into_lua(lua)?,
b"preload_workers" => t.preload_workers.into_lua(lua)?,
b"process_workers" => t.process_workers.into_lua(lua)?,
b"bizarre_retry" => t.bizarre_retry.into_lua(lua)?,
b"image_alloc" => lua.to_value_with(&t.image_alloc, SER_OPT)?,
b"image_alloc" => t.image_alloc.into_lua(lua)?,
b"image_bound" => lua.to_value_with(&t.image_bound, SER_OPT)?,
b"suppress_preload" => lua.to_value_with(&t.suppress_preload, SER_OPT)?,
b"suppress_preload" => t.suppress_preload.into_lua(lua)?,
_ => return Ok(Value::Nil),
}
.into_lua(lua)

View file

@ -0,0 +1,59 @@
use std::num::NonZeroUsize;
use anyhow::Result;
use lru::LruCache;
use parking_lot::Mutex;
use tokio::sync::mpsc;
use tracing::error;
use yazi_config::Priority;
use yazi_fs::FsHash64;
use yazi_plugin::isolate;
use yazi_shared::event::CmdCow;
use crate::{HIGH, LOW, TaskOp, TaskOps, fetch::{FetchIn, FetchOutFetch}};
pub struct Fetch {
ops: TaskOps,
tx: async_priority_channel::Sender<FetchIn, u8>,
pub loaded: Mutex<LruCache<u64, u16>>,
}
impl Fetch {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<FetchIn, u8>,
) -> Self {
Self {
ops: ops.into(),
tx,
loaded: Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap())),
}
}
pub(crate) async fn fetch(&self, task: FetchIn) -> Result<(), FetchOutFetch> {
let hashes: Vec<_> = task.targets.iter().map(|f| f.hash_u64()).collect();
let (state, err) = isolate::fetch(CmdCow::from(&task.plugin.run), task.targets).await?;
let mut loaded = self.loaded.lock();
for (_, h) in hashes.into_iter().enumerate().filter(|&(i, _)| !state.get(i)) {
loaded.get_mut(&h).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running fetcher `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, FetchOutFetch::Succ))
}
}
impl Fetch {
pub(crate) fn submit(&self, r#in: FetchIn) {
let priority = match r#in.plugin.prio {
Priority::Low => LOW,
Priority::Normal => HIGH,
Priority::High => HIGH,
};
_ = self.tx.try_send(r#in, priority);
}
}

View file

@ -0,0 +1,13 @@
use yazi_config::plugin::Fetcher;
use yazi_shared::Id;
#[derive(Debug)]
pub(crate) struct FetchIn {
pub(crate) id: Id,
pub(crate) plugin: &'static Fetcher,
pub(crate) targets: Vec<yazi_fs::File>,
}
impl FetchIn {
pub(crate) fn id(&self) -> Id { self.id }
}

View file

@ -0,0 +1 @@
yazi_macro::mod_flat!(out fetch progress r#in);

View file

@ -0,0 +1,26 @@
use crate::{Task, TaskProg};
#[derive(Debug)]
pub(crate) enum FetchOutFetch {
Succ,
Fail(String),
}
impl From<mlua::Error> for FetchOutFetch {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl FetchOutFetch {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::Fetch(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}

View file

@ -0,0 +1,32 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct FetchProg {
pub state: Option<bool>,
}
impl From<FetchProg> for TaskSummary {
fn from(value: FetchProg) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl FetchProg {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -9,19 +9,19 @@ use yazi_shared::{path::PathCow, url::{AsUrl, UrlCow, UrlLike}};
use yazi_vfs::{VfsCha, maybe_exists, provider::{self, DirEntry}, unique_file};
use super::{FileInCopy, FileInDelete, FileInHardlink, FileInLink, FileInTrash};
use crate::{LOW, NORMAL, TaskIn, TaskOp, TaskOps, ctx, file::{FileInCut, FileInDownload, FileInUpload, FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo, Transaction, Traverse}, hook::{HookInOutCopy, HookInOutCut}, ok_or_not_found, progress_or_break};
use crate::{LOW, NORMAL, TaskOp, TaskOps, ctx, file::{FileIn, FileInCut, FileInDownload, FileInUpload, FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo, Transaction, Traverse}, hook::{HookInOutCopy, HookInOutCut}, ok_or_not_found, progress_or_break};
pub(crate) struct File {
ops: TaskOps,
r#macro: async_priority_channel::Sender<TaskIn, u8>,
ops: TaskOps,
tx: async_priority_channel::Sender<FileIn, u8>,
}
impl File {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
tx: async_priority_channel::Sender<FileIn, u8>,
) -> Self {
Self { ops: ops.into(), r#macro: r#macro.clone() }
Self { ops: ops.into(), tx }
}
pub(crate) async fn copy(&self, mut task: FileInCopy) -> Result<(), FileOutCopy> {
@ -43,10 +43,10 @@ impl File {
async |task, cha| {
Ok(if cha.is_orphan() || (cha.is_link() && !task.follow) {
self.ops.out(id, FileOutCopy::New(0));
self.queue(task.into_link(), NORMAL);
self.requeue(task.into_link(), NORMAL);
} else {
self.ops.out(id, FileOutCopy::New(cha.len));
self.queue(task, LOW);
self.requeue(task, LOW);
})
},
|err| {
@ -79,7 +79,7 @@ impl File {
{
task.retry += 1;
self.ops.out(task.id, FileOutCopyDo::Log(format!("Retrying due to error: {e}")));
return Ok(self.queue(task, LOW));
return Ok(self.requeue(task, LOW));
}
Err(e) => ctx!(task, Err(e))?,
}
@ -115,10 +115,10 @@ impl File {
self.ops.out(id, FileOutCut::New(if nofollow { 0 } else { cha.len }));
if nofollow {
self.queue(task.into_link(), NORMAL);
self.requeue(task.into_link(), NORMAL);
} else {
match (cha.is_link(), reorder) {
(_, false) => self.queue(task, LOW),
(_, false) => self.requeue(task, LOW),
(true, true) => links.push(task),
(false, true) => files.push(task),
}
@ -135,14 +135,14 @@ impl File {
if !links.is_empty() {
let (tx, mut rx) = mpsc::channel(1);
for task in links {
self.queue(task.with_drop(&tx), LOW);
self.requeue(task.with_drop(&tx), LOW);
}
drop(tx);
while rx.recv().await.is_some() {}
}
for task in files {
self.queue(task, LOW);
self.requeue(task, LOW);
}
Ok(self.ops.out(id, FileOutCut::Succ))
@ -172,7 +172,7 @@ impl File {
{
task.retry += 1;
self.ops.out(task.id, FileOutCutDo::Log(format!("Retrying due to error: {e}")));
return Ok(self.queue(task, LOW));
return Ok(self.requeue(task, LOW));
}
Err(e) => ctx!(task, Err(e))?,
}
@ -186,7 +186,7 @@ impl File {
unique_file(task.to, false).await.context("Cannot determine unique destination name")?;
}
Ok(self.queue(task, NORMAL))
Ok(self.requeue(task, NORMAL))
}
pub(crate) async fn link_do(&self, task: FileInLink) -> Result<(), FileOutLink> {
@ -240,7 +240,7 @@ impl File {
},
async |task, _cha| {
self.ops.out(id, FileOutHardlink::New);
Ok(self.queue(task, NORMAL))
Ok(self.requeue(task, NORMAL))
},
|err| {
self.ops.out(id, FileOutHardlink::Deform(err));
@ -274,7 +274,7 @@ impl File {
async |_dir| Ok(()),
async |task, cha| {
self.ops.out(id, FileOutDelete::New(cha.len));
Ok(self.queue(task, NORMAL))
Ok(self.requeue(task, NORMAL))
},
|_err| {},
)
@ -294,7 +294,7 @@ impl File {
}
pub(crate) async fn trash(&self, task: FileInTrash) -> Result<(), FileOutTrash> {
Ok(self.queue(task, LOW))
Ok(self.requeue(task, LOW))
}
pub(crate) async fn trash_do(&self, task: FileInTrash) -> Result<(), FileOutTrash> {
@ -317,7 +317,7 @@ impl File {
Err(anyhow!("Failed to work on {task:?}: source of symlink doesn't exist"))?
} else {
self.ops.out(id, FileOutDownload::New(cha.len));
self.queue(task, LOW);
self.requeue(task, LOW);
})
},
|err| {
@ -364,7 +364,7 @@ impl File {
{
task.retry += 1;
self.ops.out(task.id, FileOutDownloadDo::Log(format!("Retrying due to error: {e}")));
return Ok(self.queue(task, LOW));
return Ok(self.requeue(task, LOW));
}
Err(e) => ctx!(task, Err(e))?,
}
@ -385,7 +385,7 @@ impl File {
Ok(c) if c.mtime == cha.mtime => {}
Ok(c) => {
self.ops.out(id, FileOutUpload::New(c.len));
self.queue(task, LOW);
self.requeue(task, LOW);
}
Err(e) if e.kind() == NotFound => {}
Err(e) => ctx!(task, Err(e))?,
@ -464,7 +464,12 @@ impl File {
impl File {
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.r#macro.try_send(r#in.into(), priority);
pub(crate) fn submit(&self, r#in: impl Into<FileIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
#[inline]
fn requeue(&self, r#in: impl Into<FileIn>, priority: u8) {
_ = self.tx.try_send(r#in.into().into_doable(), priority);
}
}

View file

@ -4,6 +4,81 @@ use tokio::sync::mpsc;
use yazi_fs::cha::Cha;
use yazi_shared::{CompletionToken, Id, url::UrlBuf};
#[derive(Debug)]
pub(crate) enum FileIn {
Copy(FileInCopy),
CopyDo(FileInCopy),
Cut(FileInCut),
CutDo(FileInCut),
Link(FileInLink),
LinkDo(FileInLink),
Hardlink(FileInHardlink),
HardlinkDo(FileInHardlink),
Delete(FileInDelete),
DeleteDo(FileInDelete),
Trash(FileInTrash),
TrashDo(FileInTrash),
Download(FileInDownload),
DownloadDo(FileInDownload),
Upload(FileInUpload),
UploadDo(FileInUpload),
}
impl_from_in! {
Copy(FileInCopy),
Cut(FileInCut),
Link(FileInLink),
Hardlink(FileInHardlink),
Delete(FileInDelete),
Trash(FileInTrash),
Download(FileInDownload),
Upload(FileInUpload),
}
impl FileIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Copy(r#in) => r#in.id,
Self::CopyDo(r#in) => r#in.id,
Self::Cut(r#in) => r#in.id,
Self::CutDo(r#in) => r#in.id,
Self::Link(r#in) => r#in.id,
Self::LinkDo(r#in) => r#in.id,
Self::Hardlink(r#in) => r#in.id,
Self::HardlinkDo(r#in) => r#in.id,
Self::Delete(r#in) => r#in.id,
Self::DeleteDo(r#in) => r#in.id,
Self::Trash(r#in) => r#in.id,
Self::TrashDo(r#in) => r#in.id,
Self::Download(r#in) => r#in.id,
Self::DownloadDo(r#in) => r#in.id,
Self::Upload(r#in) => r#in.id,
Self::UploadDo(r#in) => r#in.id,
}
}
pub(crate) fn into_doable(self) -> Self {
match self {
Self::Copy(r#in) => Self::CopyDo(r#in),
Self::CopyDo(_) => self,
Self::Cut(r#in) => Self::CutDo(r#in),
Self::CutDo(_) => self,
Self::Link(r#in) => Self::LinkDo(r#in),
Self::LinkDo(_) => self,
Self::Hardlink(r#in) => Self::HardlinkDo(r#in),
Self::HardlinkDo(_) => self,
Self::Delete(r#in) => Self::DeleteDo(r#in),
Self::DeleteDo(_) => self,
Self::Trash(r#in) => Self::TrashDo(r#in),
Self::TrashDo(_) => self,
Self::Download(r#in) => Self::DownloadDo(r#in),
Self::DownloadDo(_) => self,
Self::Upload(r#in) => Self::UploadDo(r#in),
Self::UploadDo(_) => self,
}
}
}
// --- Copy
#[derive(Clone, Debug)]
pub(crate) struct FileInCopy {

View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::file::FileIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(file out progress r#in transaction traverse);

View file

@ -7,16 +7,21 @@ use yazi_fs::ok_or_not_found;
use yazi_proxy::TasksProxy;
use yazi_vfs::provider;
use crate::{Ongoing, TaskOp, TaskOps, file::{FileOutCopy, FileOutCut, FileOutDelete, FileOutDownload, FileOutTrash, FileOutUpload}, hook::{HookInDelete, HookInDownload, HookInOutCopy, HookInOutCut, HookInTrash, HookInUpload}};
use crate::{Ongoing, TaskOp, TaskOps, file::{FileOutCopy, FileOutCut, FileOutDelete, FileOutDownload, FileOutTrash, FileOutUpload}, hook::{HookIn, HookInDelete, HookInDownload, HookInOutCopy, HookInOutCut, HookInTrash, HookInUpload}};
pub(crate) struct Hook {
ops: TaskOps,
ongoing: Arc<Mutex<Ongoing>>,
tx: async_priority_channel::Sender<HookIn, u8>,
}
impl Hook {
pub(crate) fn new(ops: &mpsc::UnboundedSender<TaskOp>, ongoing: &Arc<Mutex<Ongoing>>) -> Self {
Self { ops: ops.into(), ongoing: ongoing.clone() }
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
ongoing: &Arc<Mutex<Ongoing>>,
tx: async_priority_channel::Sender<HookIn, u8>,
) -> Self {
Self { ops: ops.into(), ongoing: ongoing.clone(), tx }
}
// --- File
@ -74,3 +79,10 @@ impl Hook {
self.ops.out(task.id, FileOutUpload::Clean);
}
}
impl Hook {
#[inline]
pub(crate) fn submit(&self, r#in: impl Into<HookIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

View file

@ -2,6 +2,38 @@ use yazi_shared::{Id, url::UrlBuf};
use crate::{Task, TaskProg, file::{FileInCopy, FileInCut}};
#[derive(Debug)]
pub(crate) enum HookIn {
Copy(HookInOutCopy),
Cut(HookInOutCut),
Delete(HookInDelete),
Trash(HookInTrash),
Download(HookInDownload),
Upload(HookInUpload),
}
impl_from_in!(
Copy(HookInOutCopy),
Cut(HookInOutCut),
Delete(HookInDelete),
Trash(HookInTrash),
Download(HookInDownload),
Upload(HookInUpload),
);
impl HookIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Copy(r#in) => r#in.id,
Self::Cut(r#in) => r#in.id,
Self::Delete(r#in) => r#in.id,
Self::Trash(r#in) => r#in.id,
Self::Download(r#in) => r#in.id,
Self::Upload(r#in) => r#in.id,
}
}
}
// --- Copy
#[derive(Debug)]
pub(crate) struct HookInOutCopy {

View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::hook::HookIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(hook r#in);

View file

@ -1,110 +0,0 @@
use yazi_shared::Id;
use crate::{file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload}, hook::{HookInDelete, HookInDownload, HookInOutCopy, HookInOutCut, HookInTrash, HookInUpload}, impl_from_in, plugin::PluginInEntry, prework::{PreworkInFetch, PreworkInLoad, PreworkInSize}, process::{ProcessInBg, ProcessInBlock, ProcessInOrphan}};
#[derive(Debug)]
pub(crate) enum TaskIn {
// File
FileCopy(FileInCopy),
FileCut(FileInCut),
FileLink(FileInLink),
FileHardlink(FileInHardlink),
FileDelete(FileInDelete),
FileTrash(FileInTrash),
FileDownload(FileInDownload),
FileUpload(FileInUpload),
// Plugin
PluginEntry(PluginInEntry),
// Prework
PreworkFetch(PreworkInFetch),
PreworkLoad(PreworkInLoad),
PreworkSize(PreworkInSize),
// Process
ProcessBlock(ProcessInBlock),
ProcessOrphan(ProcessInOrphan),
ProcessBg(ProcessInBg),
// Hook
HookCopy(HookInOutCopy),
HookCut(HookInOutCut),
HookDelete(HookInDelete),
HookTrash(HookInTrash),
HookDownload(HookInDownload),
HookUpload(HookInUpload),
}
impl_from_in! {
// File
FileCopy(FileInCopy), FileCut(FileInCut), FileLink(FileInLink), FileHardlink(FileInHardlink), FileDelete(FileInDelete), FileTrash(FileInTrash), FileDownload(FileInDownload), FileUpload(FileInUpload),
// Plugin
PluginEntry(PluginInEntry),
// Prework
PreworkFetch(PreworkInFetch), PreworkLoad(PreworkInLoad), PreworkSize(PreworkInSize),
// Process
ProcessBlock(ProcessInBlock), ProcessOrphan(ProcessInOrphan), ProcessBg(ProcessInBg),
// Hook
HookCopy(HookInOutCopy), HookCut(HookInOutCut), HookDelete(HookInDelete), HookTrash(HookInTrash), HookDownload(HookInDownload), HookUpload(HookInUpload),
}
impl TaskIn {
pub fn id(&self) -> Id {
match self {
// File
Self::FileCopy(r#in) => r#in.id,
Self::FileCut(r#in) => r#in.id,
Self::FileLink(r#in) => r#in.id,
Self::FileHardlink(r#in) => r#in.id,
Self::FileDelete(r#in) => r#in.id,
Self::FileTrash(r#in) => r#in.id,
Self::FileDownload(r#in) => r#in.id,
Self::FileUpload(r#in) => r#in.id,
// Plugin
Self::PluginEntry(r#in) => r#in.id,
// Prework
Self::PreworkFetch(r#in) => r#in.id,
Self::PreworkLoad(r#in) => r#in.id,
Self::PreworkSize(r#in) => r#in.id,
// Process
Self::ProcessBlock(r#in) => r#in.id,
Self::ProcessOrphan(r#in) => r#in.id,
Self::ProcessBg(r#in) => r#in.id,
// Hook
Self::HookCopy(r#in) => r#in.id,
Self::HookCut(r#in) => r#in.id,
Self::HookDelete(r#in) => r#in.id,
Self::HookTrash(r#in) => r#in.id,
Self::HookDownload(r#in) => r#in.id,
Self::HookUpload(r#in) => r#in.id,
}
}
pub fn is_hook(&self) -> bool {
match self {
// File
Self::FileCopy(_) => false,
Self::FileCut(_) => false,
Self::FileLink(_) => false,
Self::FileHardlink(_) => false,
Self::FileDelete(_) => false,
Self::FileTrash(_) => false,
Self::FileDownload(_) => false,
Self::FileUpload(_) => false,
// Plugin
Self::PluginEntry(_) => false,
// Prework
Self::PreworkFetch(_) => false,
Self::PreworkLoad(_) => false,
Self::PreworkSize(_) => false,
// Process
Self::ProcessBlock(_) => false,
Self::ProcessOrphan(_) => false,
Self::ProcessBg(_) => false,
// Hook
Self::HookCopy(_) => true,
Self::HookCut(_) => true,
Self::HookDelete(_) => true,
Self::HookTrash(_) => true,
Self::HookDownload(_) => true,
Self::HookUpload(_) => true,
}
}
}

View file

@ -1,8 +1,8 @@
mod macros;
yazi_macro::mod_pub!(file hook plugin prework process);
yazi_macro::mod_pub!(fetch file hook plugin preload process size);
yazi_macro::mod_flat!(ongoing op out progress r#in runner scheduler snap task);
yazi_macro::mod_flat!(ongoing op out progress runner scheduler snap task);
const LOW: u8 = yazi_config::Priority::Low as u8;
const NORMAL: u8 = yazi_config::Priority::Normal as u8;

View file

@ -39,17 +39,6 @@ macro_rules! progress_or_break {
};
}
#[macro_export]
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::TaskIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}
#[macro_export]
macro_rules! impl_from_out {
($($variant:ident($type:ty)),* $(,)?) => {

View file

@ -5,7 +5,7 @@ use yazi_parser::app::TaskSummary;
use yazi_shared::{CompletionToken, Id, Ids};
use super::Task;
use crate::{TaskIn, TaskProg};
use crate::{TaskProg, hook::HookIn};
#[derive(Default)]
pub struct Ongoing {
@ -23,7 +23,7 @@ impl Ongoing {
self.inner.entry(id).insert(Task::new::<T>(id, name)).into_mut()
}
pub(super) fn cancel(&mut self, id: Id) -> Option<TaskIn> {
pub(super) fn cancel(&mut self, id: Id) -> Option<HookIn> {
match self.inner.entry(id) {
Entry::Occupied(mut oe) => {
let task = oe.get_mut();

View file

@ -1,4 +1,4 @@
use crate::{Task, file::{FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo}, hook::{HookInOutCopy, HookInOutCut}, impl_from_out, plugin::PluginOutEntry, prework::{PreworkOutFetch, PreworkOutLoad, PreworkOutSize}, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}};
use crate::{Task, fetch::FetchOutFetch, file::{FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo}, hook::{HookInOutCopy, HookInOutCut}, impl_from_out, plugin::PluginOutEntry, preload::PreloadOut, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}, size::SizeOut};
#[derive(Debug)]
pub(super) enum TaskOut {
@ -19,10 +19,12 @@ pub(super) enum TaskOut {
FileUploadDo(FileOutUploadDo),
// Plugin
PluginEntry(PluginOutEntry),
// Prework
PreworkFetch(PreworkOutFetch),
PreworkLoad(PreworkOutLoad),
PreworkSize(PreworkOutSize),
// Fetch
Fetch(FetchOutFetch),
// Preload
Preload(PreloadOut),
// Size
Size(SizeOut),
// Process
ProcessBlock(ProcessOutBlock),
ProcessOrphan(ProcessOutOrphan),
@ -37,8 +39,12 @@ impl_from_out! {
FileCopy(FileOutCopy), FileCopyDo(FileOutCopyDo), FileCut(FileOutCut), FileCutDo(FileOutCutDo), FileLink(FileOutLink), FileHardlink(FileOutHardlink), FileHardlinkDo(FileOutHardlinkDo), FileDelete(FileOutDelete), FileDeleteDo(FileOutDeleteDo), FileTrash(FileOutTrash), FileDownload(FileOutDownload), FileDownloadDo(FileOutDownloadDo), FileUpload(FileOutUpload), FileUploadDo(FileOutUploadDo),
// Plugin
PluginEntry(PluginOutEntry),
// Prework
PreworkFetch(PreworkOutFetch), PreworkLoad(PreworkOutLoad), PreworkSize(PreworkOutSize),
// Fetch
Fetch(FetchOutFetch),
// Preload
Preload(PreloadOut),
// Size
Size(SizeOut),
// Process
ProcessBlock(ProcessOutBlock), ProcessOrphan(ProcessOutOrphan), ProcessBg(ProcessOutBg),
// Hook
@ -66,9 +72,9 @@ impl TaskOut {
// Plugin
Self::PluginEntry(out) => out.reduce(task),
// Prework
Self::PreworkFetch(out) => out.reduce(task),
Self::PreworkLoad(out) => out.reduce(task),
Self::PreworkSize(out) => out.reduce(task),
Self::Fetch(out) => out.reduce(task),
Self::Preload(out) => out.reduce(task),
Self::Size(out) => out.reduce(task),
// Process
Self::ProcessBlock(out) => out.reduce(task),
Self::ProcessOrphan(out) => out.reduce(task),

View file

@ -1,6 +1,22 @@
use yazi_parser::app::PluginOpt;
use yazi_shared::Id;
#[derive(Debug)]
pub(crate) enum PluginIn {
Entry(PluginInEntry),
}
impl_from_in!(Entry(PluginInEntry));
impl PluginIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Entry(r#in) => r#in.id,
}
}
}
// --- Entry
#[derive(Debug)]
pub(crate) struct PluginInEntry {
pub(crate) id: Id,

View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::plugin::PluginIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(out plugin progress r#in);

View file

@ -3,26 +3,22 @@ use tokio::sync::mpsc;
use yazi_plugin::isolate;
use super::PluginInEntry;
use crate::{HIGH, TaskIn, TaskOp, TaskOps, plugin::PluginOutEntry};
use crate::{TaskOp, TaskOps, plugin::{PluginIn, PluginOutEntry}};
pub(crate) struct Plugin {
ops: TaskOps,
r#macro: async_priority_channel::Sender<TaskIn, u8>,
ops: TaskOps,
tx: async_priority_channel::Sender<PluginIn, u8>,
}
impl Plugin {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
tx: async_priority_channel::Sender<PluginIn, u8>,
) -> Self {
Self { ops: ops.into(), r#macro: r#macro.clone() }
Self { ops: ops.into(), tx }
}
pub(crate) async fn entry(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> {
Ok(self.queue(task, HIGH))
}
pub(crate) async fn entry_do(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> {
isolate::entry(task.opt).await?;
Ok(self.ops.out(task.id, PluginOutEntry::Succ))
}
@ -30,7 +26,7 @@ impl Plugin {
impl Plugin {
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.r#macro.try_send(r#in.into(), priority);
pub(crate) fn submit(&self, r#in: impl Into<PluginIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

View file

@ -0,0 +1,13 @@
use yazi_config::plugin::Preloader;
use yazi_shared::Id;
#[derive(Clone, Debug)]
pub(crate) struct PreloadIn {
pub(crate) id: Id,
pub(crate) plugin: &'static Preloader,
pub(crate) target: yazi_fs::File,
}
impl PreloadIn {
pub(crate) fn id(&self) -> Id { self.id }
}

View file

@ -0,0 +1 @@
yazi_macro::mod_flat!(out preload progress r#in);

View file

@ -0,0 +1,26 @@
use crate::{Task, TaskProg};
#[derive(Debug)]
pub(crate) enum PreloadOut {
Succ,
Fail(String),
}
impl From<mlua::Error> for PreloadOut {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl PreloadOut {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::Preload(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}

View file

@ -0,0 +1,67 @@
use std::num::NonZeroUsize;
use anyhow::Result;
use lru::LruCache;
use parking_lot::Mutex;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::error;
use yazi_config::Priority;
use yazi_fs::FsHash64;
use yazi_plugin::isolate;
use crate::{HIGH, LOW, NORMAL, TaskOp, TaskOps, preload::{PreloadIn, PreloadOut}};
pub struct Preload {
ops: TaskOps,
tx: async_priority_channel::Sender<PreloadIn, u8>,
pub loaded: Mutex<LruCache<u64, u16>>,
pub loading: Mutex<LruCache<u64, CancellationToken>>,
}
impl Preload {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<PreloadIn, u8>,
) -> Self {
Self {
ops: ops.into(),
tx,
loaded: Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap())),
loading: Mutex::new(LruCache::new(NonZeroUsize::new(256).unwrap())),
}
}
pub(crate) async fn preload(&self, task: PreloadIn) -> Result<(), PreloadOut> {
let ct = CancellationToken::new();
if let Some(ct) = self.loading.lock().put(task.target.url.hash_u64(), ct.clone()) {
ct.cancel();
}
let hash = task.target.hash_u64();
let (ok, err) = isolate::preload(&task.plugin.run, task.target, ct).await?;
if !ok {
self.loaded.lock().get_mut(&hash).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running preloader `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, PreloadOut::Succ))
}
}
impl Preload {
pub(crate) fn submit(&self, r#in: PreloadIn) {
let priority = match r#in.plugin.prio {
Priority::Low => LOW,
Priority::Normal => NORMAL,
Priority::High => HIGH,
};
_ = self.tx.try_send(r#in.into(), priority);
}
}

View file

@ -0,0 +1,32 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreloadProg {
pub state: Option<bool>,
}
impl From<PreloadProg> for TaskSummary {
fn from(value: PreloadProg) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl PreloadProg {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -1,25 +0,0 @@
use std::sync::Arc;
use yazi_config::plugin::{Fetcher, Preloader};
use yazi_shared::{Id, Throttle, url::UrlBuf};
#[derive(Debug)]
pub(crate) struct PreworkInFetch {
pub(crate) id: Id,
pub(crate) plugin: &'static Fetcher,
pub(crate) targets: Vec<yazi_fs::File>,
}
#[derive(Clone, Debug)]
pub(crate) struct PreworkInLoad {
pub(crate) id: Id,
pub(crate) plugin: &'static Preloader,
pub(crate) target: yazi_fs::File,
}
#[derive(Debug)]
pub(crate) struct PreworkInSize {
pub(crate) id: Id,
pub(crate) target: UrlBuf,
pub(crate) throttle: Arc<Throttle<(UrlBuf, u64)>>,
}

View file

@ -1 +0,0 @@
yazi_macro::mod_flat!(out prework progress r#in);

View file

@ -1,70 +0,0 @@
use crate::{Task, TaskProg};
// --- Fetch
#[derive(Debug)]
pub(crate) enum PreworkOutFetch {
Succ,
Fail(String),
}
impl From<mlua::Error> for PreworkOutFetch {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl PreworkOutFetch {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::PreworkFetch(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}
// --- Load
#[derive(Debug)]
pub(crate) enum PreworkOutLoad {
Succ,
Fail(String),
}
impl From<mlua::Error> for PreworkOutLoad {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl PreworkOutLoad {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::PreworkLoad(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}
// --- Size
#[derive(Debug)]
pub(crate) enum PreworkOutSize {
Done,
}
impl PreworkOutSize {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::PreworkSize(prog) = &mut task.prog else { return };
match self {
Self::Done => {
prog.done = true;
}
}
}
}

View file

@ -1,123 +0,0 @@
use std::num::NonZeroUsize;
use anyhow::Result;
use hashbrown::{HashMap, HashSet};
use lru::LruCache;
use parking_lot::{Mutex, RwLock};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::error;
use yazi_config::Priority;
use yazi_fs::{FilesOp, FsHash64};
use yazi_plugin::isolate;
use yazi_shared::{event::CmdCow, url::{UrlBuf, UrlLike}};
use yazi_vfs::provider;
use super::{PreworkInFetch, PreworkInLoad, PreworkInSize};
use crate::{HIGH, NORMAL, TaskIn, TaskOp, TaskOps, prework::{PreworkOutFetch, PreworkOutLoad, PreworkOutSize}};
pub struct Prework {
ops: TaskOps,
r#macro: async_priority_channel::Sender<TaskIn, u8>,
pub loaded: Mutex<LruCache<u64, u32>>,
pub loading: Mutex<LruCache<u64, CancellationToken>>,
pub sizing: RwLock<HashSet<UrlBuf>>,
}
impl Prework {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
) -> Self {
Self {
ops: ops.into(),
r#macro: r#macro.clone(),
loaded: Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap())),
loading: Mutex::new(LruCache::new(NonZeroUsize::new(256).unwrap())),
sizing: Default::default(),
}
}
pub(crate) async fn fetch(&self, task: PreworkInFetch) -> Result<(), PreworkOutFetch> {
match task.plugin.prio {
Priority::Low => Ok(self.queue(task, NORMAL)),
Priority::Normal => Ok(self.queue(task, HIGH)),
Priority::High => self.fetch_do(task).await,
}
}
pub(crate) async fn fetch_do(&self, task: PreworkInFetch) -> Result<(), PreworkOutFetch> {
let hashes: Vec<_> = task.targets.iter().map(|f| f.hash_u64()).collect();
let (state, err) = isolate::fetch(CmdCow::from(&task.plugin.run), task.targets).await?;
let mut loaded = self.loaded.lock();
for (_, h) in hashes.into_iter().enumerate().filter(|&(i, _)| !state.get(i)) {
loaded.get_mut(&h).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running fetcher `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, PreworkOutFetch::Succ))
}
pub(crate) async fn load(&self, task: PreworkInLoad) -> Result<(), PreworkOutLoad> {
match task.plugin.prio {
Priority::Low => Ok(self.queue(task, NORMAL)),
Priority::Normal => Ok(self.queue(task, HIGH)),
Priority::High => self.load_do(task).await,
}
}
pub(crate) async fn load_do(&self, task: PreworkInLoad) -> Result<(), PreworkOutLoad> {
let ct = CancellationToken::new();
if let Some(ct) = self.loading.lock().put(task.target.url.hash_u64(), ct.clone()) {
ct.cancel();
}
let hash = task.target.hash_u64();
let (ok, err) = isolate::preload(&task.plugin.run, task.target, ct).await?;
if !ok {
self.loaded.lock().get_mut(&hash).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running preloader `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, PreworkOutLoad::Succ))
}
pub(crate) async fn size(&self, task: PreworkInSize) -> Result<(), PreworkOutSize> {
self.size_do(task).await
}
pub(crate) async fn size_do(&self, task: PreworkInSize) -> Result<(), PreworkOutSize> {
let length = provider::calculate(&task.target).await.unwrap_or(0);
task.throttle.done((task.target, length), |buf| {
{
let mut loading = self.sizing.write();
for (path, _) in &buf {
loading.remove(path);
}
}
let parent = buf[0].0.parent().unwrap();
FilesOp::Size(
parent.into(),
HashMap::from_iter(buf.into_iter().map(|(u, s)| (u.urn().into(), s))),
)
.emit();
});
Ok(self.ops.out(task.id, PreworkOutSize::Done))
}
}
impl Prework {
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.r#macro.try_send(r#in.into(), priority);
}
}

View file

@ -1,95 +0,0 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
// --- Fetch
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreworkProgFetch {
pub state: Option<bool>,
}
impl From<PreworkProgFetch> for TaskSummary {
fn from(value: PreworkProgFetch) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl PreworkProgFetch {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}
// --- Load
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreworkProgLoad {
pub state: Option<bool>,
}
impl From<PreworkProgLoad> for TaskSummary {
fn from(value: PreworkProgLoad) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl PreworkProgLoad {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}
// --- Size
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreworkProgSize {
pub done: bool,
}
impl From<PreworkProgSize> for TaskSummary {
fn from(value: PreworkProgSize) -> Self {
Self {
total: 1,
success: value.done as u32,
failed: 0,
percent: value.percent().map(Into::into),
}
}
}
impl PreworkProgSize {
pub fn cooked(self) -> bool { self.done }
pub fn running(self) -> bool { !self.done }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { false }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -4,6 +4,25 @@ use yazi_shared::{CompletionToken, Id, url::UrlCow};
use super::ShellOpt;
#[derive(Debug)]
pub(crate) enum ProcessIn {
Block(ProcessInBlock),
Orphan(ProcessInOrphan),
Bg(ProcessInBg),
}
impl_from_in!(Block(ProcessInBlock), Orphan(ProcessInOrphan), Bg(ProcessInBg));
impl ProcessIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Block(r#in) => r#in.id,
Self::Orphan(r#in) => r#in.id,
Self::Bg(r#in) => r#in.id,
}
}
}
// --- Block
#[derive(Debug)]
pub(crate) struct ProcessInBlock {

View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::process::ProcessIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(out process progress r#in shell);

View file

@ -4,14 +4,20 @@ use yazi_binding::Permit;
use yazi_proxy::{AppProxy, HIDER, NotifyProxy};
use super::{ProcessInBg, ProcessInBlock, ProcessInOrphan, ShellOpt};
use crate::{TaskOp, TaskOps, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}};
use crate::{TaskOp, TaskOps, process::{ProcessIn, ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}};
pub(crate) struct Process {
ops: TaskOps,
tx: async_priority_channel::Sender<ProcessIn, u8>,
}
impl Process {
pub(crate) fn new(ops: &mpsc::UnboundedSender<TaskOp>) -> Self { Self { ops: ops.into() } }
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<ProcessIn, u8>,
) -> Self {
Self { ops: ops.into(), tx }
}
pub(crate) async fn block(&self, task: ProcessInBlock) -> Result<(), ProcessOutBlock> {
let _permit = Permit::new(HIDER.acquire().await.unwrap(), AppProxy::resume());
@ -85,3 +91,10 @@ impl Process {
Ok(self.ops.out(task.id, ProcessOutBg::Succ))
}
}
impl Process {
#[inline]
pub(crate) fn submit(&self, r#in: impl Into<ProcessIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

View file

@ -1,7 +1,7 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
use crate::{file::{FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, impl_from_prog, plugin::PluginProgEntry, prework::{PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}};
use crate::{fetch::FetchProg, file::{FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, impl_from_prog, plugin::PluginProgEntry, preload::PreloadProg, process::{ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}, size::SizeProg};
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(tag = "kind")]
@ -17,10 +17,12 @@ pub enum TaskProg {
FileUpload(FileProgUpload),
// Plugin
PluginEntry(PluginProgEntry),
// Prework
PreworkFetch(PreworkProgFetch),
PreworkLoad(PreworkProgLoad),
PreworkSize(PreworkProgSize),
// Fetch
Fetch(FetchProg),
// Preload
Preload(PreloadProg),
// Size
Size(SizeProg),
// Process
ProcessBlock(ProcessProgBlock),
ProcessOrphan(ProcessProgOrphan),
@ -32,8 +34,12 @@ impl_from_prog! {
FileCopy(FileProgCopy), FileCut(FileProgCut), FileLink(FileProgLink), FileHardlink(FileProgHardlink), FileDelete(FileProgDelete), FileTrash(FileProgTrash), FileDownload(FileProgDownload), FileUpload(FileProgUpload),
// Plugin
PluginEntry(PluginProgEntry),
// Prework
PreworkFetch(PreworkProgFetch), PreworkLoad(PreworkProgLoad), PreworkSize(PreworkProgSize),
// Fetch
Fetch(FetchProg),
// Preload
Preload(PreloadProg),
// Size
Size(SizeProg),
// Process
ProcessBlock(ProcessProgBlock), ProcessOrphan(ProcessProgOrphan), ProcessBg(ProcessProgBg),
}
@ -53,9 +59,9 @@ impl From<TaskProg> for TaskSummary {
// Plugin
TaskProg::PluginEntry(p) => p.into(),
// Prework
TaskProg::PreworkFetch(p) => p.into(),
TaskProg::PreworkLoad(p) => p.into(),
TaskProg::PreworkSize(p) => p.into(),
TaskProg::Fetch(p) => p.into(),
TaskProg::Preload(p) => p.into(),
TaskProg::Size(p) => p.into(),
// Process
TaskProg::ProcessBlock(p) => p.into(),
TaskProg::ProcessOrphan(p) => p.into(),
@ -79,9 +85,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.cooked(),
// Prework
Self::PreworkFetch(p) => p.cooked(),
Self::PreworkLoad(p) => p.cooked(),
Self::PreworkSize(p) => p.cooked(),
Self::Fetch(p) => p.cooked(),
Self::Preload(p) => p.cooked(),
Self::Size(p) => p.cooked(),
// Process
Self::ProcessBlock(p) => p.cooked(),
Self::ProcessOrphan(p) => p.cooked(),
@ -103,9 +109,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.running(),
// Prework
Self::PreworkFetch(p) => p.running(),
Self::PreworkLoad(p) => p.running(),
Self::PreworkSize(p) => p.running(),
Self::Fetch(p) => p.running(),
Self::Preload(p) => p.running(),
Self::Size(p) => p.running(),
// Process
Self::ProcessBlock(p) => p.running(),
Self::ProcessOrphan(p) => p.running(),
@ -127,9 +133,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.success(),
// Prework
Self::PreworkFetch(p) => p.success(),
Self::PreworkLoad(p) => p.success(),
Self::PreworkSize(p) => p.success(),
Self::Fetch(p) => p.success(),
Self::Preload(p) => p.success(),
Self::Size(p) => p.success(),
// Process
Self::ProcessBlock(p) => p.success(),
Self::ProcessOrphan(p) => p.success(),
@ -151,9 +157,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.failed(),
// Prework
Self::PreworkFetch(p) => p.failed(),
Self::PreworkLoad(p) => p.failed(),
Self::PreworkSize(p) => p.failed(),
Self::Fetch(p) => p.failed(),
Self::Preload(p) => p.failed(),
Self::Size(p) => p.failed(),
// Process
Self::ProcessBlock(p) => p.failed(),
Self::ProcessOrphan(p) => p.failed(),
@ -175,9 +181,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.cleaned(),
// Prework
Self::PreworkFetch(p) => p.cleaned(),
Self::PreworkLoad(p) => p.cleaned(),
Self::PreworkSize(p) => p.cleaned(),
Self::Fetch(p) => p.cleaned(),
Self::Preload(p) => p.cleaned(),
Self::Size(p) => p.cleaned(),
// Process
Self::ProcessBlock(p) => p.cleaned(),
Self::ProcessOrphan(p) => p.cleaned(),
@ -199,9 +205,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.percent(),
// Prework
Self::PreworkFetch(p) => p.percent(),
Self::PreworkLoad(p) => p.percent(),
Self::PreworkSize(p) => p.percent(),
Self::Fetch(p) => p.percent(),
Self::Preload(p) => p.percent(),
Self::Size(p) => p.percent(),
// Process
Self::ProcessBlock(p) => p.percent(),
Self::ProcessOrphan(p) => p.percent(),
@ -223,9 +229,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(_) => true,
// Prework
Self::PreworkFetch(_) => false,
Self::PreworkLoad(_) => false,
Self::PreworkSize(_) => false,
Self::Fetch(_) => false,
Self::Preload(_) => false,
Self::Size(_) => false,
// Process
Self::ProcessBlock(_) => true,
Self::ProcessOrphan(_) => true,

View file

@ -1,76 +1,294 @@
use std::sync::Arc;
use crate::{TaskIn, TaskOut, file::File, hook::Hook, plugin::Plugin, prework::Prework, process::Process};
use parking_lot::Mutex;
use tokio::{select, sync::mpsc, task::JoinHandle};
use yazi_config::YAZI;
use crate::{LOW, Ongoing, TaskOp, TaskOps, TaskOut, fetch::{Fetch, FetchIn}, file::{File, FileIn}, hook::{Hook, HookIn}, plugin::{Plugin, PluginIn}, preload::{Preload, PreloadIn}, process::{Process, ProcessIn}, size::{Size, SizeIn}};
#[derive(Clone)]
pub struct Runner {
pub(super) file: Arc<File>,
pub(super) plugin: Arc<Plugin>,
pub prework: Arc<Prework>,
pub fetch: Arc<Fetch>,
pub preload: Arc<Preload>,
pub size: Arc<Size>,
pub(super) process: Arc<Process>,
pub(super) hook: Arc<Hook>,
pub ops: TaskOps,
pub ongoing: Arc<Mutex<Ongoing>>,
}
impl Runner {
pub(super) async fn micro(&self, r#in: TaskIn) -> Result<(), TaskOut> {
pub(super) fn make() -> (Self, Vec<JoinHandle<()>>) {
let (file_tx, file_rx) = async_priority_channel::unbounded();
let (plugin_tx, plugin_rx) = async_priority_channel::unbounded();
let (fetch_tx, fetch_rx) = async_priority_channel::unbounded();
let (preload_tx, preload_rx) = async_priority_channel::unbounded();
let (size_tx, size_rx) = async_priority_channel::unbounded();
let (process_tx, process_rx) = async_priority_channel::unbounded();
let (hook_tx, hook_rx) = async_priority_channel::unbounded();
let (op_tx, op_rx) = mpsc::unbounded_channel();
let ongoing = Arc::new(Mutex::new(Ongoing::default()));
let me = Self {
file: Arc::new(File::new(&op_tx, file_tx)),
plugin: Arc::new(Plugin::new(&op_tx, plugin_tx)),
fetch: Arc::new(Fetch::new(&op_tx, fetch_tx)),
preload: Arc::new(Preload::new(&op_tx, preload_tx)),
size: Arc::new(Size::new(&op_tx, size_tx)),
process: Arc::new(Process::new(&op_tx, process_tx)),
hook: Arc::new(Hook::new(&op_tx, &ongoing, hook_tx)),
ops: TaskOps(op_tx),
ongoing,
};
let handles = []
.into_iter()
.chain((0..YAZI.tasks.file_workers).map(|_| me.file(file_rx.clone())))
.chain((0..YAZI.tasks.plugin_workers).map(|_| me.plugin(plugin_rx.clone())))
.chain((0..YAZI.tasks.fetch_workers).map(|_| me.fetch(fetch_rx.clone())))
.chain((0..YAZI.tasks.preload_workers).map(|_| me.preload(preload_rx.clone())))
.chain((0..3).map(|_| me.size(size_rx.clone())))
.chain((0..YAZI.tasks.process_workers).map(|_| me.process(process_rx.clone())))
.chain((0..3).map(|_| me.hook(hook_rx.clone())))
.chain([me.op(op_rx)])
.collect();
(me, handles)
}
fn file(&self, rx: async_priority_channel::Receiver<FileIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.file_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn file_do(&self, r#in: FileIn) -> Result<(), TaskOut> {
match r#in {
// File
TaskIn::FileCopy(r#in) => self.file.copy(r#in).await.map_err(Into::into),
TaskIn::FileCut(r#in) => self.file.cut(r#in).await.map_err(Into::into),
TaskIn::FileLink(r#in) => self.file.link(r#in).await.map_err(Into::into),
TaskIn::FileHardlink(r#in) => self.file.hardlink(r#in).await.map_err(Into::into),
TaskIn::FileDelete(r#in) => self.file.delete(r#in).await.map_err(Into::into),
TaskIn::FileTrash(r#in) => self.file.trash(r#in).await.map_err(Into::into),
TaskIn::FileDownload(r#in) => self.file.download(r#in).await.map_err(Into::into),
TaskIn::FileUpload(r#in) => self.file.upload(r#in).await.map_err(Into::into),
// Plugin
TaskIn::PluginEntry(r#in) => self.plugin.entry(r#in).await.map_err(Into::into),
// Prework
TaskIn::PreworkFetch(r#in) => self.prework.fetch(r#in).await.map_err(Into::into),
TaskIn::PreworkLoad(r#in) => self.prework.load(r#in).await.map_err(Into::into),
TaskIn::PreworkSize(r#in) => self.prework.size(r#in).await.map_err(Into::into),
// Process
TaskIn::ProcessBlock(r#in) => self.process.block(r#in).await.map_err(Into::into),
TaskIn::ProcessOrphan(r#in) => self.process.orphan(r#in).await.map_err(Into::into),
TaskIn::ProcessBg(r#in) => self.process.bg(r#in).await.map_err(Into::into),
// Hook
TaskIn::HookCopy(r#in) => Ok(self.hook.copy(r#in).await),
TaskIn::HookCut(r#in) => Ok(self.hook.cut(r#in).await),
TaskIn::HookDelete(r#in) => Ok(self.hook.delete(r#in).await),
TaskIn::HookTrash(r#in) => Ok(self.hook.trash(r#in).await),
TaskIn::HookDownload(r#in) => Ok(self.hook.download(r#in).await),
TaskIn::HookUpload(r#in) => Ok(self.hook.upload(r#in).await),
FileIn::Copy(r#in) => self.file.copy(r#in).await.map_err(Into::into),
FileIn::CopyDo(r#in) => self.file.copy_do(r#in).await.map_err(Into::into),
FileIn::Cut(r#in) => self.file.cut(r#in).await.map_err(Into::into),
FileIn::CutDo(r#in) => self.file.cut_do(r#in).await.map_err(Into::into),
FileIn::Link(r#in) => self.file.link(r#in).await.map_err(Into::into),
FileIn::LinkDo(r#in) => self.file.link_do(r#in).await.map_err(Into::into),
FileIn::Hardlink(r#in) => self.file.hardlink(r#in).await.map_err(Into::into),
FileIn::HardlinkDo(r#in) => self.file.hardlink_do(r#in).await.map_err(Into::into),
FileIn::Delete(r#in) => self.file.delete(r#in).await.map_err(Into::into),
FileIn::DeleteDo(r#in) => self.file.delete_do(r#in).await.map_err(Into::into),
FileIn::Trash(r#in) => self.file.trash(r#in).await.map_err(Into::into),
FileIn::TrashDo(r#in) => self.file.trash_do(r#in).await.map_err(Into::into),
FileIn::Download(r#in) => self.file.download(r#in).await.map_err(Into::into),
FileIn::DownloadDo(r#in) => self.file.download_do(r#in).await.map_err(Into::into),
FileIn::Upload(r#in) => self.file.upload(r#in).await.map_err(Into::into),
FileIn::UploadDo(r#in) => self.file.upload_do(r#in).await.map_err(Into::into),
}
}
pub(super) async fn r#macro(&self, r#in: TaskIn) -> Result<(), TaskOut> {
fn plugin(&self, rx: async_priority_channel::Receiver<PluginIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.plugin_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn plugin_do(&self, r#in: PluginIn) -> Result<(), TaskOut> {
match r#in {
// File
TaskIn::FileCopy(r#in) => self.file.copy_do(r#in).await.map_err(Into::into),
TaskIn::FileCut(r#in) => self.file.cut_do(r#in).await.map_err(Into::into),
TaskIn::FileLink(r#in) => self.file.link_do(r#in).await.map_err(Into::into),
TaskIn::FileHardlink(r#in) => self.file.hardlink_do(r#in).await.map_err(Into::into),
TaskIn::FileDelete(r#in) => self.file.delete_do(r#in).await.map_err(Into::into),
TaskIn::FileTrash(r#in) => self.file.trash_do(r#in).await.map_err(Into::into),
TaskIn::FileDownload(r#in) => self.file.download_do(r#in).await.map_err(Into::into),
TaskIn::FileUpload(r#in) => self.file.upload_do(r#in).await.map_err(Into::into),
// Plugin
TaskIn::PluginEntry(r#in) => self.plugin.entry_do(r#in).await.map_err(Into::into),
// Prework
TaskIn::PreworkFetch(r#in) => self.prework.fetch_do(r#in).await.map_err(Into::into),
TaskIn::PreworkLoad(r#in) => self.prework.load_do(r#in).await.map_err(Into::into),
TaskIn::PreworkSize(r#in) => self.prework.size_do(r#in).await.map_err(Into::into),
// Process
TaskIn::ProcessBlock(_in) => unreachable!(),
TaskIn::ProcessOrphan(_in) => unreachable!(),
TaskIn::ProcessBg(_in) => unreachable!(),
// Hook
TaskIn::HookCopy(_in) => unreachable!(),
TaskIn::HookCut(_in) => unreachable!(),
TaskIn::HookDelete(_in) => unreachable!(),
TaskIn::HookTrash(_in) => unreachable!(),
TaskIn::HookDownload(_in) => unreachable!(),
TaskIn::HookUpload(_in) => unreachable!(),
PluginIn::Entry(r#in) => self.plugin.entry(r#in).await.map_err(Into::into),
}
}
fn fetch(&self, rx: async_priority_channel::Receiver<FetchIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.fetch_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn fetch_do(&self, r#in: FetchIn) -> Result<(), TaskOut> {
self.fetch.fetch(r#in).await.map_err(Into::into)
}
fn preload(&self, rx: async_priority_channel::Receiver<PreloadIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.preload_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn preload_do(&self, r#in: PreloadIn) -> Result<(), TaskOut> {
self.preload.preload(r#in).await.map_err(Into::into)
}
fn size(&self, rx: async_priority_channel::Receiver<SizeIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.size_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn size_do(&self, r#in: SizeIn) -> Result<(), TaskOut> {
self.size.size(r#in).await.map_err(Into::into)
}
fn process(&self, rx: async_priority_channel::Receiver<ProcessIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.process_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn process_do(&self, r#in: ProcessIn) -> Result<(), TaskOut> {
match r#in {
ProcessIn::Block(r#in) => self.process.block(r#in).await.map_err(Into::into),
ProcessIn::Orphan(r#in) => self.process.orphan(r#in).await.map_err(Into::into),
ProcessIn::Bg(r#in) => self.process.bg(r#in).await.map_err(Into::into),
}
}
fn hook(&self, rx: async_priority_channel::Receiver<HookIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
if !me.ongoing.lock().exists(id) {
continue;
}
me.hook_do(r#in).await;
}
}
})
}
async fn hook_do(&self, r#in: HookIn) {
match r#in {
HookIn::Copy(r#in) => self.hook.copy(r#in).await,
HookIn::Cut(r#in) => self.hook.cut(r#in).await,
HookIn::Delete(r#in) => self.hook.delete(r#in).await,
HookIn::Trash(r#in) => self.hook.trash(r#in).await,
HookIn::Download(r#in) => self.hook.download(r#in).await,
HookIn::Upload(r#in) => self.hook.upload(r#in).await,
}
}
fn op(&self, mut rx: mpsc::UnboundedReceiver<TaskOp>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
while let Some(op) = rx.recv().await {
let mut ongoing = me.ongoing.lock();
let Some(task) = ongoing.get_mut(op.id) else { continue };
op.out.reduce(task);
if !task.prog.cooked() && task.done.completed() != Some(false) {
continue; // Not cooked yet, also not canceled
} else if task.prog.cleaned() == Some(false) {
continue; // Failed to clean up
} else if let Some(hook) = task.hook.take() {
me.hook.submit(hook, LOW);
} else {
ongoing.fulfill(op.id);
}
}
})
}
}

View file

@ -1,60 +1,32 @@
use std::{sync::Arc, time::Duration};
use std::{ops::Deref, sync::Arc, time::Duration};
use parking_lot::Mutex;
use tokio::{select, sync::mpsc::{self, UnboundedReceiver}, task::JoinHandle};
use tokio::task::JoinHandle;
use yazi_config::{YAZI, plugin::{Fetcher, Preloader}};
use yazi_parser::{app::PluginOpt, tasks::ProcessOpenOpt};
use yazi_shared::{CompletionToken, Id, Throttle, url::{UrlBuf, UrlLike}};
use super::{Ongoing, TaskOp};
use crate::{HIGH, LOW, NORMAL, Runner, TaskIn, TaskOps, file::{File, FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload, FileOutCopy, FileOutCut, FileOutDownload, FileOutHardlink, FileOutUpload, FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, hook::{Hook, HookInDelete, HookInDownload, HookInTrash, HookInUpload}, plugin::{Plugin, PluginInEntry, PluginProgEntry}, prework::{Prework, PreworkInFetch, PreworkInLoad, PreworkInSize, PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{Process, ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}};
use crate::{HIGH, LOW, NORMAL, Runner, fetch::{FetchIn, FetchProg}, file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload, FileOutCopy, FileOutCut, FileOutDownload, FileOutHardlink, FileOutUpload, FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, hook::{HookInDelete, HookInDownload, HookInTrash, HookInUpload}, plugin::{PluginInEntry, PluginProgEntry}, preload::{PreloadIn, PreloadProg}, process::{ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}, size::{SizeIn, SizeProg}};
pub struct Scheduler {
ops: TaskOps,
pub runner: Runner,
micro: async_priority_channel::Sender<TaskIn, u8>,
handles: Vec<JoinHandle<()>>,
pub ongoing: Arc<Mutex<Ongoing>>,
pub runner: Runner,
handles: Vec<JoinHandle<()>>,
}
impl Deref for Scheduler {
type Target = Runner;
fn deref(&self) -> &Self::Target { &self.runner }
}
impl Scheduler {
pub fn serve() -> Self {
let (op_tx, op_rx) = mpsc::unbounded_channel();
let (micro_tx, micro_rx) = async_priority_channel::unbounded();
let (macro_tx, macro_rx) = async_priority_channel::unbounded();
let ongoing = Arc::new(Mutex::new(Ongoing::default()));
let runner = Runner {
file: Arc::new(File::new(&op_tx, &macro_tx)),
plugin: Arc::new(Plugin::new(&op_tx, &macro_tx)),
prework: Arc::new(Prework::new(&op_tx, &macro_tx)),
process: Arc::new(Process::new(&op_tx)),
hook: Arc::new(Hook::new(&op_tx, &ongoing)),
};
let mut scheduler = Self {
ops: TaskOps(op_tx),
runner,
micro: micro_tx,
handles: Vec::with_capacity(
YAZI.tasks.micro_workers as usize + YAZI.tasks.macro_workers as usize + 1,
),
ongoing,
};
for _ in 0..YAZI.tasks.micro_workers {
scheduler.handles.push(scheduler.schedule_micro(micro_rx.clone()));
}
for _ in 0..YAZI.tasks.macro_workers {
scheduler.handles.push(scheduler.schedule_macro(micro_rx.clone(), macro_rx.clone()));
}
scheduler.handle_ops(op_rx);
scheduler
let (runner, handles) = Runner::make();
Self { runner, handles }
}
pub fn cancel(&self, id: Id) -> bool {
if let Some(hook) = self.ongoing.lock().cancel(id) {
self.micro.try_send(hook, HIGH).ok();
self.hook.submit(hook, HIGH);
return false;
}
@ -78,7 +50,7 @@ impl Scheduler {
}
let follow = !from.scheme().covariant(to.scheme());
self.queue(
self.file.submit(
FileInCut {
id: task.id,
from,
@ -105,7 +77,7 @@ impl Scheduler {
}
let follow = follow || !from.scheme().covariant(to.scheme());
self.queue(
self.file.submit(
FileInCopy {
id: task.id,
from,
@ -124,7 +96,7 @@ impl Scheduler {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<FileProgLink>(format!("Link {} to {}", from.display(), to.display()));
self.queue(
self.file.submit(
FileInLink {
id: task.id,
from,
@ -156,7 +128,7 @@ impl Scheduler {
.out(task.id, FileOutHardlink::Fail("Cannot hardlink directory into itself".to_owned()));
}
self.queue(FileInHardlink { id: task.id, from, to, force, cha: None, follow }, LOW);
self.file.submit(FileInHardlink { id: task.id, from, to, force, cha: None, follow }, LOW);
}
pub fn file_delete(&self, target: UrlBuf) {
@ -164,7 +136,7 @@ impl Scheduler {
let task = ongoing.add::<FileProgDelete>(format!("Delete {}", target.display()));
task.set_hook(HookInDelete { id: task.id, target: target.clone() });
self.queue(FileInDelete { id: task.id, target, cha: None }, LOW);
self.file.submit(FileInDelete { id: task.id, target, cha: None }, LOW);
}
pub fn file_trash(&self, target: UrlBuf) {
@ -172,7 +144,7 @@ impl Scheduler {
let task = ongoing.add::<FileProgTrash>(format!("Trash {}", target.display()));
task.set_hook(HookInTrash { id: task.id, target: target.clone() });
self.queue(FileInTrash { id: task.id, target }, LOW);
self.file.submit(FileInTrash { id: task.id, target }, LOW);
}
pub fn file_download(&self, url: UrlBuf) -> CompletionToken {
@ -181,7 +153,7 @@ impl Scheduler {
if url.kind().is_remote() {
task.set_hook(HookInDownload { id: task.id });
self.queue(
self.file.submit(
FileInDownload { id: task.id, url, cha: None, retry: 0, done: task.done.clone() },
LOW,
);
@ -203,7 +175,7 @@ impl Scheduler {
};
task.set_hook(HookInUpload { id: task.id, target: url.clone() });
self.queue(
self.file.submit(
FileInUpload { id: task.id, url, cha: None, cache: None, done: task.done.clone() },
LOW,
);
@ -213,7 +185,7 @@ impl Scheduler {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PluginProgEntry>(format!("Run micro plugin `{}`", opt.id));
self.queue(PluginInEntry { id: task.id, opt }, NORMAL);
self.plugin.submit(PluginInEntry { id: task.id, opt }, NORMAL);
}
pub fn fetch_paged(
@ -222,13 +194,13 @@ impl Scheduler {
targets: Vec<yazi_fs::File>,
) -> CompletionToken {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PreworkProgFetch>(format!(
let task = ongoing.add::<FetchProg>(format!(
"Run fetcher `{}` with {} target(s)",
fetcher.run.name,
targets.len()
));
self.queue(PreworkInFetch { id: task.id, plugin: fetcher, targets }, NORMAL);
self.fetch.submit(FetchIn { id: task.id, plugin: fetcher, targets });
task.done.clone()
}
@ -248,10 +220,10 @@ impl Scheduler {
pub fn preload_paged(&self, preloader: &'static Preloader, target: &yazi_fs::File) {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PreworkProgLoad>(format!("Run preloader `{}`", preloader.run.name));
let task = ongoing.add::<PreloadProg>(format!("Run preloader `{}`", preloader.run.name));
let target = target.clone();
self.queue(PreworkInLoad { id: task.id, plugin: preloader, target }, NORMAL);
self.preload.submit(PreloadIn { id: task.id, plugin: preloader, target });
}
pub fn prework_size(&self, targets: Vec<&UrlBuf>) {
@ -259,12 +231,11 @@ impl Scheduler {
let mut ongoing = self.ongoing.lock();
for target in targets {
let task =
ongoing.add::<PreworkProgSize>(format!("Calculate the size of {}", target.display()));
let task = ongoing.add::<SizeProg>(format!("Calculate the size of {}", target.display()));
let target = target.clone();
let throttle = throttle.clone();
self.queue(PreworkInSize { id: task.id, target, throttle }, NORMAL);
self.size.submit(SizeIn { id: task.id, target, throttle }, NORMAL);
}
}
@ -293,12 +264,15 @@ impl Scheduler {
if opt.block {
self
.queue(ProcessInBlock { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL);
.process
.submit(ProcessInBlock { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL);
} else if opt.orphan {
self
.queue(ProcessInOrphan { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL);
self.process.submit(
ProcessInOrphan { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args },
NORMAL,
);
} else {
self.queue(
self.process.submit(
ProcessInBg {
id: task.id,
cwd: opt.cwd,
@ -310,104 +284,4 @@ impl Scheduler {
);
};
}
fn schedule_micro(&self, rx: async_priority_channel::Receiver<TaskIn, u8>) -> JoinHandle<()> {
let ops = self.ops.clone();
let runner = self.runner.clone();
let ongoing = self.ongoing.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = ongoing.lock().get_token(id) else {
continue;
};
let result = if r#in.is_hook() {
runner.micro(r#in).await
} else {
select! {
r = runner.micro(r#in) => r,
false = token.future() => Ok(())
}
};
if let Err(out) = result {
ops.out(id, out);
}
}
}
})
}
fn schedule_macro(
&self,
micro: async_priority_channel::Receiver<TaskIn, u8>,
r#macro: async_priority_channel::Receiver<TaskIn, u8>,
) -> JoinHandle<()> {
let ops = self.ops.clone();
let runner = self.runner.clone();
let ongoing = self.ongoing.clone();
tokio::spawn(async move {
loop {
let (r#in, micro) = select! {
Ok((r#in, _)) = micro.recv() => (r#in, true),
Ok((r#in, _)) = r#macro.recv() => (r#in, false),
};
let id = r#in.id();
let Some(token) = ongoing.lock().get_token(id) else {
continue;
};
let result = if r#in.is_hook() {
if micro { runner.micro(r#in).await } else { runner.r#macro(r#in).await }
} else if micro {
select! {
r = runner.micro(r#in) => r,
false = token.future() => Ok(()),
}
} else {
select! {
r = runner.r#macro(r#in) => r,
false = token.future() => Ok(()),
}
};
if let Err(out) = result {
ops.out(id, out);
}
}
})
}
fn handle_ops(&self, mut rx: UnboundedReceiver<TaskOp>) -> JoinHandle<()> {
let micro = self.micro.clone();
let ongoing = self.ongoing.clone();
tokio::spawn(async move {
while let Some(op) = rx.recv().await {
let mut ongoing = ongoing.lock();
let Some(task) = ongoing.get_mut(op.id) else { continue };
op.out.reduce(task);
if !task.prog.cooked() && task.done.completed() != Some(false) {
continue; // Not cooked yet, also not canceled
} else if task.prog.cleaned() == Some(false) {
continue; // Failed to clean up
} else if let Some(hook) = task.hook.take() {
micro.try_send(hook, LOW).ok();
} else {
ongoing.fulfill(op.id);
}
}
})
}
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.micro.try_send(r#in.into(), priority);
}
}

View file

@ -0,0 +1,14 @@
use std::sync::Arc;
use yazi_shared::{Id, Throttle, url::UrlBuf};
#[derive(Debug)]
pub(crate) struct SizeIn {
pub(crate) id: Id,
pub(crate) target: UrlBuf,
pub(crate) throttle: Arc<Throttle<(UrlBuf, u64)>>,
}
impl SizeIn {
pub(crate) fn id(&self) -> Id { self.id }
}

View file

@ -0,0 +1 @@
yazi_macro::mod_flat!(out progress r#in size);

View file

@ -0,0 +1,17 @@
use crate::{Task, TaskProg};
#[derive(Debug)]
pub(crate) enum SizeOut {
Done,
}
impl SizeOut {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::Size(prog) = &mut task.prog else { return };
match self {
Self::Done => {
prog.done = true;
}
}
}
}

View file

@ -0,0 +1,32 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct SizeProg {
pub done: bool,
}
impl From<SizeProg> for TaskSummary {
fn from(value: SizeProg) -> Self {
Self {
total: 1,
success: value.done as u32,
failed: 0,
percent: value.percent().map(Into::into),
}
}
}
impl SizeProg {
pub fn cooked(self) -> bool { self.done }
pub fn running(self) -> bool { !self.done }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { false }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -0,0 +1,54 @@
use anyhow::Result;
use hashbrown::{HashMap, HashSet};
use parking_lot::RwLock;
use tokio::sync::mpsc;
use yazi_fs::FilesOp;
use yazi_shared::url::{UrlBuf, UrlLike};
use yazi_vfs::provider;
use super::SizeIn;
use crate::{TaskOp, TaskOps, size::SizeOut};
pub struct Size {
ops: TaskOps,
tx: async_priority_channel::Sender<SizeIn, u8>,
pub sizing: RwLock<HashSet<UrlBuf>>,
}
impl Size {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<SizeIn, u8>,
) -> Self {
Self { ops: ops.into(), tx, sizing: Default::default() }
}
pub(crate) async fn size(&self, task: SizeIn) -> Result<(), SizeOut> {
let length = provider::calculate(&task.target).await.unwrap_or(0);
task.throttle.done((task.target, length), |buf| {
{
let mut loading = self.sizing.write();
for (path, _) in &buf {
loading.remove(path);
}
}
let parent = buf[0].0.parent().unwrap();
FilesOp::Size(
parent.into(),
HashMap::from_iter(buf.into_iter().map(|(u, s)| (u.urn().into(), s))),
)
.emit();
});
Ok(self.ops.out(task.id, SizeOut::Done))
}
}
impl Size {
#[inline]
pub(crate) fn submit(&self, r#in: impl Into<SizeIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

View file

@ -1,14 +1,14 @@
use tokio::sync::mpsc;
use yazi_shared::{CompletionToken, Id};
use crate::{TaskIn, TaskProg};
use crate::{TaskProg, hook::HookIn};
#[derive(Debug)]
pub struct Task {
pub id: Id,
pub name: String,
pub(crate) prog: TaskProg,
pub(crate) hook: Option<TaskIn>,
pub(crate) hook: Option<HookIn>,
pub done: CompletionToken,
pub logs: String,
@ -41,5 +41,5 @@ impl Task {
}
}
pub(super) fn set_hook(&mut self, hook: impl Into<TaskIn>) { self.hook = Some(hook.into()); }
pub(super) fn set_hook(&mut self, hook: impl Into<HookIn>) { self.hook = Some(hook.into()); }
}

View file

@ -209,7 +209,7 @@ impl Cmd {
let key = parts.next().expect("at least one part");
let val = parts.next().map_or(Data::Boolean(true), Data::from);
Ok((DataKey::from(key.to_owned()), val))
Ok((key.to_owned().into(), val))
})
.collect()
}