Compare commits
1 Commits
v0.1.1
...
369e83e52d
| Author | SHA1 | Date | |
|---|---|---|---|
|
369e83e52d |
11
CHANGELOG.md
11
CHANGELOG.md
@@ -6,17 +6,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.1.1] - 2025-11-24
|
||||
|
||||
### Added
|
||||
- replace channel with semaphore for better performance
|
||||
|
||||
### Fixed
|
||||
- *(deps)* update all dependencies
|
||||
|
||||
### Other
|
||||
- *(deps)* update rust crate tracing to v0.1.41
|
||||
|
||||
## [0.1.0] - 2025-08-08
|
||||
|
||||
### Added
|
||||
|
||||
188
Cargo.lock
generated
188
Cargo.lock
generated
@@ -4,9 +4,9 @@ version = 4
|
||||
|
||||
[[package]]
|
||||
name = "anyhow"
|
||||
version = "1.0.100"
|
||||
version = "1.0.71"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
|
||||
checksum = "9c7d0618f0e0b7e8ff11427422b64564d5fb0be1940354bfe2e0529b18a9d9b8"
|
||||
|
||||
[[package]]
|
||||
name = "autocfg"
|
||||
@@ -45,10 +45,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.177"
|
||||
name = "hermit-abi"
|
||||
version = "0.2.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2874a2af47a2325c2001a6e6fad9b16a53b802102b528163885171cf92b15976"
|
||||
checksum = "ee512640fe35acbfb4bb779db6f0d80704c2cacfa2e39b601ef3e3f47d1ae4c7"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.146"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f92be4933c13fd498862a9e02a3055f8a8d9c039ce33db97306fd5a6caa7f29b"
|
||||
|
||||
[[package]]
|
||||
name = "lock_api"
|
||||
@@ -68,18 +77,18 @@ checksum = "b06a4cde4c0f271a446782e3eff8de789548ce57dbc8eca9292c27f4a42004b4"
|
||||
|
||||
[[package]]
|
||||
name = "mio"
|
||||
version = "1.1.0"
|
||||
version = "0.8.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "69d83b0086dc8ecf3ce9ae2874b2d1290252e2a30720bea58a5c6639b0092873"
|
||||
checksum = "927a765cd3fc26206e66b296465fa9d3e5ab003e651c1b3c060e7956d96b19d2"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"wasi",
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "noworkers"
|
||||
version = "0.1.0"
|
||||
version = "0.0.4"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"tokio",
|
||||
@@ -87,6 +96,16 @@ dependencies = [
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_cpus"
|
||||
version = "1.15.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0fac9e2da13b5eb447a6ce3d392f23a29d8694bff781bf03a16cd9ac8697593b"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "once_cell"
|
||||
version = "1.18.0"
|
||||
@@ -113,7 +132,7 @@ dependencies = [
|
||||
"libc",
|
||||
"redox_syscall",
|
||||
"smallvec",
|
||||
"windows-targets 0.48.0",
|
||||
"windows-targets",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -172,12 +191,12 @@ checksum = "a507befe795404456341dfab10cef66ead4c041f62b8b11bbb92bffe5d0953e0"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.6.1"
|
||||
version = "0.4.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "17129e116933cf371d018bb80ae557e889637989d8638274fb25622827b03881"
|
||||
checksum = "64a4a911eed85daf18834cfaa86a79b7d266ff93ff5ba14005426219480ed662"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys 0.60.2",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -193,26 +212,28 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tokio"
|
||||
version = "1.48.0"
|
||||
version = "1.28.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408"
|
||||
checksum = "94d7b1cfd2aa4011f2de74c2c4c63665e27a71006b0a192dcd2710272e73dfa2"
|
||||
dependencies = [
|
||||
"autocfg",
|
||||
"bytes",
|
||||
"libc",
|
||||
"mio",
|
||||
"num_cpus",
|
||||
"parking_lot",
|
||||
"pin-project-lite",
|
||||
"signal-hook-registry",
|
||||
"socket2",
|
||||
"tokio-macros",
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-macros"
|
||||
version = "2.6.0"
|
||||
version = "2.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5"
|
||||
checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -221,9 +242,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tokio-util"
|
||||
version = "0.7.17"
|
||||
version = "0.7.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2efa149fe76073d6e8fd97ef4f4eca7b67f599660115591483572e406e165594"
|
||||
checksum = "66a539a9ad6d5d281510d5bd368c973d636c02dbf8a67300bfb6b950696ad7df"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
@@ -234,10 +255,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tracing"
|
||||
version = "0.1.41"
|
||||
version = "0.1.37"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0"
|
||||
checksum = "8ce8c33a8d48bd45d624a6e523445fd21ec13d3653cd51f681abf67418f54eb8"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"log",
|
||||
"pin-project-lite",
|
||||
"tracing-attributes",
|
||||
@@ -246,9 +268,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tracing-attributes"
|
||||
version = "0.1.30"
|
||||
version = "0.1.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903"
|
||||
checksum = "0f57e3ca2a01450b1a921183a9c9cbfda207fd822cef4ccb00a65402cbba7a74"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -257,9 +279,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tracing-core"
|
||||
version = "0.1.34"
|
||||
version = "0.1.31"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678"
|
||||
checksum = "0955b8137a1df6f1a2e9a37d8a6656291ff0297c1a97c24e0d8425fe2312f79a"
|
||||
dependencies = [
|
||||
"once_cell",
|
||||
]
|
||||
@@ -277,27 +299,34 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
|
||||
|
||||
[[package]]
|
||||
name = "windows-link"
|
||||
version = "0.2.1"
|
||||
name = "winapi"
|
||||
version = "0.3.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
|
||||
|
||||
[[package]]
|
||||
name = "windows-sys"
|
||||
version = "0.60.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
|
||||
checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
|
||||
dependencies = [
|
||||
"windows-targets 0.53.5",
|
||||
"winapi-i686-pc-windows-gnu",
|
||||
"winapi-x86_64-pc-windows-gnu",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-sys"
|
||||
version = "0.61.2"
|
||||
name = "winapi-i686-pc-windows-gnu"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
|
||||
checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
|
||||
|
||||
[[package]]
|
||||
name = "winapi-x86_64-pc-windows-gnu"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
|
||||
|
||||
[[package]]
|
||||
name = "windows-sys"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
|
||||
dependencies = [
|
||||
"windows-link",
|
||||
"windows-targets",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -306,30 +335,13 @@ version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7b1eb6f0cd7c80c79759c929114ef071b87354ce476d9d94271031c0497adfd5"
|
||||
dependencies = [
|
||||
"windows_aarch64_gnullvm 0.48.0",
|
||||
"windows_aarch64_msvc 0.48.0",
|
||||
"windows_i686_gnu 0.48.0",
|
||||
"windows_i686_msvc 0.48.0",
|
||||
"windows_x86_64_gnu 0.48.0",
|
||||
"windows_x86_64_gnullvm 0.48.0",
|
||||
"windows_x86_64_msvc 0.48.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-targets"
|
||||
version = "0.53.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
|
||||
dependencies = [
|
||||
"windows-link",
|
||||
"windows_aarch64_gnullvm 0.53.1",
|
||||
"windows_aarch64_msvc 0.53.1",
|
||||
"windows_i686_gnu 0.53.1",
|
||||
"windows_i686_gnullvm",
|
||||
"windows_i686_msvc 0.53.1",
|
||||
"windows_x86_64_gnu 0.53.1",
|
||||
"windows_x86_64_gnullvm 0.53.1",
|
||||
"windows_x86_64_msvc 0.53.1",
|
||||
"windows_aarch64_gnullvm",
|
||||
"windows_aarch64_msvc",
|
||||
"windows_i686_gnu",
|
||||
"windows_i686_msvc",
|
||||
"windows_x86_64_gnu",
|
||||
"windows_x86_64_gnullvm",
|
||||
"windows_x86_64_msvc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -338,86 +350,38 @@ version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "91ae572e1b79dba883e0d315474df7305d12f569b400fcf90581b06062f7e1bc"
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_gnullvm"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_msvc"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b2ef27e0d7bdfcfc7b868b317c1d32c641a6fe4629c171b8928c7b08d98d7cf3"
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_msvc"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnu"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "622a1962a7db830d6fd0a69683c80a18fda201879f0f447f065a3b7467daa241"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnu"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnullvm"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_msvc"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4542c6e364ce21bf45d69fdd2a8e455fa38d316158cfd43b3ac1c5b1b19f8e00"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_msvc"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnu"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ca2b8a661f7628cbd23440e50b05d705db3686f894fc9580820623656af974b1"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnu"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnullvm"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7896dbc1f41e08872e9d5e8f8baa8fdd2677f29468c4e156210174edc7f7b953"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnullvm"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_msvc"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_msvc"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
|
||||
|
||||
@@ -3,7 +3,7 @@ members = ["crates/*"]
|
||||
resolver = "2"
|
||||
|
||||
[workspace.package]
|
||||
version = "0.1.1"
|
||||
version = "0.1.0"
|
||||
license = "MIT"
|
||||
|
||||
[workspace.dependencies]
|
||||
|
||||
@@ -18,10 +18,10 @@ async fn main() -> anyhow::Result<()> {
|
||||
workers
|
||||
.add(move |_cancel| async move {
|
||||
println!("Task {} starting...", i);
|
||||
|
||||
|
||||
// Simulate some async work
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(100 * (i as u64 + 1))).await;
|
||||
|
||||
|
||||
println!("Task {} completed!", i);
|
||||
Ok(())
|
||||
})
|
||||
@@ -36,4 +36,4 @@ async fn main() -> anyhow::Result<()> {
|
||||
println!("\nAll tasks completed successfully!");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -17,15 +17,15 @@ async fn main() -> anyhow::Result<()> {
|
||||
// Simulate a large dataset
|
||||
let dataset: Vec<i32> = (1..=100).collect();
|
||||
let total_items = dataset.len();
|
||||
|
||||
|
||||
println!("Processing {} items in parallel...\n", total_items);
|
||||
|
||||
// Example 1: Simple parallel map
|
||||
example_parallel_map(dataset.clone()).await?;
|
||||
|
||||
|
||||
// Example 2: Pipeline with multiple stages
|
||||
example_pipeline(dataset.clone()).await?;
|
||||
|
||||
|
||||
// Example 3: Batch processing
|
||||
example_batch_processing(dataset).await?;
|
||||
|
||||
@@ -35,149 +35,147 @@ async fn main() -> anyhow::Result<()> {
|
||||
async fn example_parallel_map(dataset: Vec<i32>) -> anyhow::Result<()> {
|
||||
println!("1. Simple Parallel Map");
|
||||
println!("----------------------");
|
||||
|
||||
|
||||
let mut workers = Workers::new();
|
||||
|
||||
|
||||
// Use system CPU count for optimal parallelism
|
||||
workers.with_limit_to_system_cpus();
|
||||
|
||||
|
||||
let results = Arc::new(tokio::sync::Mutex::new(Vec::new()));
|
||||
let processed = Arc::new(AtomicUsize::new(0));
|
||||
let total = dataset.len();
|
||||
|
||||
|
||||
let start = Instant::now();
|
||||
|
||||
|
||||
for item in dataset {
|
||||
let results = results.clone();
|
||||
let processed = processed.clone();
|
||||
|
||||
|
||||
workers
|
||||
.add(move |_cancel| async move {
|
||||
// Simulate CPU-intensive processing
|
||||
let result = expensive_computation(item).await?;
|
||||
|
||||
|
||||
// Store result
|
||||
let mut res = results.lock().await;
|
||||
res.push(result);
|
||||
|
||||
|
||||
// Update progress
|
||||
let count = processed.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
if count.is_multiple_of(10) {
|
||||
if count % 10 == 0 {
|
||||
println!(" Processed {}/{} items", count, total);
|
||||
}
|
||||
|
||||
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
|
||||
workers.wait().await?;
|
||||
|
||||
|
||||
let results = results.lock().await;
|
||||
let sum: i32 = results.iter().sum();
|
||||
|
||||
|
||||
println!(" Completed in {}ms", start.elapsed().as_millis());
|
||||
println!(" Sum of results: {}\n", sum);
|
||||
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn example_pipeline(dataset: Vec<i32>) -> anyhow::Result<()> {
|
||||
println!("2. Multi-Stage Pipeline");
|
||||
println!("-----------------------");
|
||||
|
||||
|
||||
// Stage 1: Transform
|
||||
let stage1_output = Arc::new(tokio::sync::Mutex::new(Vec::new()));
|
||||
|
||||
|
||||
let mut stage1 = Workers::new();
|
||||
stage1.with_limit(4);
|
||||
|
||||
|
||||
println!(" Stage 1: Transforming data...");
|
||||
for item in dataset {
|
||||
let output = stage1_output.clone();
|
||||
|
||||
|
||||
stage1
|
||||
.add(move |_| async move {
|
||||
let transformed = item * 2;
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
|
||||
|
||||
let mut out = output.lock().await;
|
||||
out.push(transformed);
|
||||
|
||||
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
|
||||
stage1.wait().await?;
|
||||
|
||||
|
||||
// Stage 2: Filter and aggregate
|
||||
let stage2_output = Arc::new(tokio::sync::Mutex::new(Vec::new()));
|
||||
|
||||
|
||||
let mut stage2 = Workers::new();
|
||||
stage2.with_limit(2);
|
||||
|
||||
|
||||
println!(" Stage 2: Filtering and aggregating...");
|
||||
let stage1_data = stage1_output.lock().await.clone();
|
||||
|
||||
|
||||
for chunk in stage1_data.chunks(10) {
|
||||
let chunk = chunk.to_vec();
|
||||
let output = stage2_output.clone();
|
||||
|
||||
|
||||
stage2
|
||||
.add(move |_| async move {
|
||||
// Filter even numbers and sum
|
||||
let filtered_sum: i32 = chunk.iter().filter(|&&x| x % 2 == 0).sum();
|
||||
|
||||
let filtered_sum: i32 = chunk.iter()
|
||||
.filter(|&&x| x % 2 == 0)
|
||||
.sum();
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
|
||||
|
||||
let mut out = output.lock().await;
|
||||
out.push(filtered_sum);
|
||||
|
||||
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
|
||||
stage2.wait().await?;
|
||||
|
||||
|
||||
let final_results = stage2_output.lock().await;
|
||||
let total: i32 = final_results.iter().sum();
|
||||
|
||||
|
||||
println!(" Pipeline result: {}\n", total);
|
||||
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn example_batch_processing(dataset: Vec<i32>) -> anyhow::Result<()> {
|
||||
println!("3. Batch Processing");
|
||||
println!("-------------------");
|
||||
|
||||
|
||||
let batch_size = 20;
|
||||
let batches: Vec<Vec<i32>> = dataset
|
||||
.chunks(batch_size)
|
||||
.map(|chunk| chunk.to_vec())
|
||||
.collect();
|
||||
|
||||
println!(
|
||||
" Processing {} batches of {} items each",
|
||||
batches.len(),
|
||||
batch_size
|
||||
);
|
||||
|
||||
|
||||
println!(" Processing {} batches of {} items each", batches.len(), batch_size);
|
||||
|
||||
let mut workers = Workers::new();
|
||||
workers.with_limit(3); // Process 3 batches concurrently
|
||||
|
||||
|
||||
let results = Arc::new(tokio::sync::Mutex::new(Vec::new()));
|
||||
let start = Instant::now();
|
||||
|
||||
|
||||
for (batch_idx, batch) in batches.into_iter().enumerate() {
|
||||
let results = results.clone();
|
||||
|
||||
|
||||
workers
|
||||
.add(move |cancel| async move {
|
||||
println!(" Batch {} started", batch_idx);
|
||||
|
||||
|
||||
// Process batch with cancellation support
|
||||
tokio::select! {
|
||||
batch_result = process_batch(batch, batch_idx) => {
|
||||
@@ -202,42 +200,38 @@ async fn example_batch_processing(dataset: Vec<i32>) -> anyhow::Result<()> {
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
|
||||
workers.wait().await?;
|
||||
|
||||
|
||||
let results = results.lock().await;
|
||||
let total_processed: usize = results.iter().sum();
|
||||
|
||||
println!(
|
||||
" Processed {} items in {}ms\n",
|
||||
total_processed,
|
||||
start.elapsed().as_millis()
|
||||
);
|
||||
|
||||
|
||||
println!(" Processed {} items in {}ms\n", total_processed, start.elapsed().as_millis());
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn expensive_computation(n: i32) -> anyhow::Result<i32> {
|
||||
// Simulate CPU-intensive work
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
|
||||
|
||||
// Some complex calculation
|
||||
let result = (n * n) + (n / 2) - 1;
|
||||
|
||||
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn process_batch(batch: Vec<i32>, _batch_idx: usize) -> anyhow::Result<usize> {
|
||||
// Simulate batch processing
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
|
||||
// Process each item in the batch
|
||||
let processed_count = batch.len();
|
||||
|
||||
|
||||
// In a real scenario, you might:
|
||||
// - Write to a database
|
||||
// - Send to an API
|
||||
// - Transform and save to files
|
||||
|
||||
|
||||
Ok(processed_count)
|
||||
}
|
||||
}
|
||||
@@ -1,62 +0,0 @@
|
||||
#![feature(random)]
|
||||
use std::{random, sync::atomic::AtomicUsize};
|
||||
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
static SLOW_IN_PROGRESS: AtomicUsize = AtomicUsize::new(0);
|
||||
static FAST_IN_PROGRESS: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
println!(
|
||||
"slow: {}, fast: {}",
|
||||
SLOW_IN_PROGRESS.load(std::sync::atomic::Ordering::Relaxed),
|
||||
FAST_IN_PROGRESS.load(std::sync::atomic::Ordering::Relaxed)
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
}
|
||||
});
|
||||
let mut workers = noworkers::Workers::new();
|
||||
|
||||
workers.with_limit(30);
|
||||
|
||||
for _ in 0..1000 {
|
||||
let range: u16 = random::random(..);
|
||||
if range < (u16::MAX / 4) {
|
||||
workers.add(slow).await?;
|
||||
continue;
|
||||
}
|
||||
|
||||
workers.add(fast).await?;
|
||||
}
|
||||
|
||||
workers.wait().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn fast(_cancel: CancellationToken) -> anyhow::Result<()> {
|
||||
FAST_IN_PROGRESS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
// println!("{}: running fast", now());
|
||||
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
|
||||
|
||||
FAST_IN_PROGRESS.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
|
||||
Ok(())
|
||||
}
|
||||
async fn slow(_cancel: CancellationToken) -> anyhow::Result<()> {
|
||||
SLOW_IN_PROGRESS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
// println!("{}: running slow", now());
|
||||
tokio::time::sleep(std::time::Duration::from_secs(3)).await;
|
||||
// println!("{}: completed slow", now());
|
||||
SLOW_IN_PROGRESS.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn now() -> u128 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis()
|
||||
}
|
||||
@@ -191,10 +191,7 @@
|
||||
|
||||
use std::{future::Future, sync::Arc};
|
||||
|
||||
use tokio::{
|
||||
sync::{Mutex, OwnedSemaphorePermit, Semaphore},
|
||||
task::JoinHandle,
|
||||
};
|
||||
use tokio::{sync::Mutex, task::JoinHandle};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
/// Extension traits for common patterns.
|
||||
@@ -325,7 +322,8 @@ enum WorkerLimit {
|
||||
#[default]
|
||||
NoLimit,
|
||||
Amount {
|
||||
done: Arc<tokio::sync::Semaphore>,
|
||||
queue: tokio::sync::mpsc::Sender<()>,
|
||||
done: Arc<Mutex<tokio::sync::mpsc::Receiver<()>>>,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -333,21 +331,18 @@ impl WorkerLimit {
|
||||
pub async fn queue_worker(&self) -> WorkerGuard {
|
||||
match self {
|
||||
WorkerLimit::NoLimit => {}
|
||||
WorkerLimit::Amount { done } => {
|
||||
WorkerLimit::Amount { queue, .. } => {
|
||||
// Queue work, if the channel is limited, we will block until there is enough room
|
||||
let permit = done
|
||||
.clone()
|
||||
.acquire_owned()
|
||||
queue
|
||||
.send(())
|
||||
.await
|
||||
.expect("to be able to acquire permit");
|
||||
|
||||
return WorkerGuard {
|
||||
_permit: Some(permit),
|
||||
};
|
||||
.expect("tried to queue work on a closed worker channel");
|
||||
}
|
||||
}
|
||||
|
||||
WorkerGuard { _permit: None }
|
||||
WorkerGuard {
|
||||
limit: self.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -358,7 +353,24 @@ impl WorkerLimit {
|
||||
///
|
||||
/// This type is not directly constructible by users.
|
||||
pub struct WorkerGuard {
|
||||
_permit: Option<OwnedSemaphorePermit>,
|
||||
limit: WorkerLimit,
|
||||
}
|
||||
|
||||
impl Drop for WorkerGuard {
|
||||
fn drop(&mut self) {
|
||||
match &self.limit {
|
||||
WorkerLimit::NoLimit => { /* no limit on dequeue */ }
|
||||
WorkerLimit::Amount { done, .. } => {
|
||||
let done = done.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut done = done.lock().await;
|
||||
|
||||
// dequeue an item, leave room for the next
|
||||
done.recv().await
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Workers {
|
||||
@@ -530,8 +542,11 @@ impl Workers {
|
||||
/// # }
|
||||
/// ```
|
||||
pub fn with_limit(&mut self, limit: usize) -> &mut Self {
|
||||
let (tx, rx) = tokio::sync::mpsc::channel(limit);
|
||||
|
||||
self.limit = WorkerLimit::Amount {
|
||||
done: Arc::new(Semaphore::new(limit)),
|
||||
queue: tx,
|
||||
done: Arc::new(Mutex::new(rx)),
|
||||
};
|
||||
self
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user