3 Commits

Author SHA1 Message Date
5b6df5d596 chore(release): v0.1.1 (#9)
All checks were successful
continuous-integration/drone/push Build is passing
continuous-integration/drone/tag Build is passing
chore(release): 0.1.1

Co-authored-by: cuddle-please <bot@cuddle.sh>
Reviewed-on: #9
2025-11-25 09:03:15 +01:00
38946f77c6 feat: replace channel with semaphore for better performance
All checks were successful
continuous-integration/drone/push Build is passing
2025-11-24 23:25:11 +01:00
2b35c06d88 fix(deps): update all dependencies
All checks were successful
continuous-integration/drone/push Build is passing
2025-11-13 03:25:48 +01:00
7 changed files with 263 additions and 162 deletions

View File

@@ -6,6 +6,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased] ## [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 ## [0.1.0] - 2025-08-08
### Added ### Added

173
Cargo.lock generated
View File

@@ -4,9 +4,9 @@ version = 4
[[package]] [[package]]
name = "anyhow" name = "anyhow"
version = "1.0.71" version = "1.0.100"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c7d0618f0e0b7e8ff11427422b64564d5fb0be1940354bfe2e0529b18a9d9b8" checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
[[package]] [[package]]
name = "autocfg" name = "autocfg"
@@ -44,20 +44,11 @@ version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
[[package]]
name = "hermit-abi"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee512640fe35acbfb4bb779db6f0d80704c2cacfa2e39b601ef3e3f47d1ae4c7"
dependencies = [
"libc",
]
[[package]] [[package]]
name = "libc" name = "libc"
version = "0.2.146" version = "0.2.177"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f92be4933c13fd498862a9e02a3055f8a8d9c039ce33db97306fd5a6caa7f29b" checksum = "2874a2af47a2325c2001a6e6fad9b16a53b802102b528163885171cf92b15976"
[[package]] [[package]]
name = "lock_api" name = "lock_api"
@@ -77,13 +68,13 @@ checksum = "b06a4cde4c0f271a446782e3eff8de789548ce57dbc8eca9292c27f4a42004b4"
[[package]] [[package]]
name = "mio" name = "mio"
version = "0.8.8" version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "927a765cd3fc26206e66b296465fa9d3e5ab003e651c1b3c060e7956d96b19d2" checksum = "69d83b0086dc8ecf3ce9ae2874b2d1290252e2a30720bea58a5c6639b0092873"
dependencies = [ dependencies = [
"libc", "libc",
"wasi", "wasi",
"windows-sys", "windows-sys 0.61.2",
] ]
[[package]] [[package]]
@@ -96,16 +87,6 @@ dependencies = [
"tracing", "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]] [[package]]
name = "once_cell" name = "once_cell"
version = "1.18.0" version = "1.18.0"
@@ -132,7 +113,7 @@ dependencies = [
"libc", "libc",
"redox_syscall", "redox_syscall",
"smallvec", "smallvec",
"windows-targets", "windows-targets 0.48.0",
] ]
[[package]] [[package]]
@@ -191,12 +172,12 @@ checksum = "a507befe795404456341dfab10cef66ead4c041f62b8b11bbb92bffe5d0953e0"
[[package]] [[package]]
name = "socket2" name = "socket2"
version = "0.4.9" version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "64a4a911eed85daf18834cfaa86a79b7d266ff93ff5ba14005426219480ed662" checksum = "17129e116933cf371d018bb80ae557e889637989d8638274fb25622827b03881"
dependencies = [ dependencies = [
"libc", "libc",
"winapi", "windows-sys 0.60.2",
] ]
[[package]] [[package]]
@@ -212,28 +193,26 @@ dependencies = [
[[package]] [[package]]
name = "tokio" name = "tokio"
version = "1.28.2" version = "1.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94d7b1cfd2aa4011f2de74c2c4c63665e27a71006b0a192dcd2710272e73dfa2" checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408"
dependencies = [ dependencies = [
"autocfg",
"bytes", "bytes",
"libc", "libc",
"mio", "mio",
"num_cpus",
"parking_lot", "parking_lot",
"pin-project-lite", "pin-project-lite",
"signal-hook-registry", "signal-hook-registry",
"socket2", "socket2",
"tokio-macros", "tokio-macros",
"windows-sys", "windows-sys 0.61.2",
] ]
[[package]] [[package]]
name = "tokio-macros" name = "tokio-macros"
version = "2.1.0" version = "2.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -242,9 +221,9 @@ dependencies = [
[[package]] [[package]]
name = "tokio-util" name = "tokio-util"
version = "0.7.15" version = "0.7.17"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "66a539a9ad6d5d281510d5bd368c973d636c02dbf8a67300bfb6b950696ad7df" checksum = "2efa149fe76073d6e8fd97ef4f4eca7b67f599660115591483572e406e165594"
dependencies = [ dependencies = [
"bytes", "bytes",
"futures-core", "futures-core",
@@ -298,34 +277,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
[[package]] [[package]]
name = "winapi" name = "windows-link"
version = "0.3.9" version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
dependencies = [
"winapi-i686-pc-windows-gnu",
"winapi-x86_64-pc-windows-gnu",
]
[[package]]
name = "winapi-i686-pc-windows-gnu"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
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]] [[package]]
name = "windows-sys" name = "windows-sys"
version = "0.48.0" version = "0.60.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
dependencies = [ dependencies = [
"windows-targets", "windows-targets 0.53.5",
]
[[package]]
name = "windows-sys"
version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
"windows-link",
] ]
[[package]] [[package]]
@@ -334,13 +306,30 @@ version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7b1eb6f0cd7c80c79759c929114ef071b87354ce476d9d94271031c0497adfd5" checksum = "7b1eb6f0cd7c80c79759c929114ef071b87354ce476d9d94271031c0497adfd5"
dependencies = [ dependencies = [
"windows_aarch64_gnullvm", "windows_aarch64_gnullvm 0.48.0",
"windows_aarch64_msvc", "windows_aarch64_msvc 0.48.0",
"windows_i686_gnu", "windows_i686_gnu 0.48.0",
"windows_i686_msvc", "windows_i686_msvc 0.48.0",
"windows_x86_64_gnu", "windows_x86_64_gnu 0.48.0",
"windows_x86_64_gnullvm", "windows_x86_64_gnullvm 0.48.0",
"windows_x86_64_msvc", "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",
] ]
[[package]] [[package]]
@@ -349,38 +338,86 @@ version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "91ae572e1b79dba883e0d315474df7305d12f569b400fcf90581b06062f7e1bc" checksum = "91ae572e1b79dba883e0d315474df7305d12f569b400fcf90581b06062f7e1bc"
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
[[package]] [[package]]
name = "windows_aarch64_msvc" name = "windows_aarch64_msvc"
version = "0.48.0" version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b2ef27e0d7bdfcfc7b868b317c1d32c641a6fe4629c171b8928c7b08d98d7cf3" checksum = "b2ef27e0d7bdfcfc7b868b317c1d32c641a6fe4629c171b8928c7b08d98d7cf3"
[[package]]
name = "windows_aarch64_msvc"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
[[package]] [[package]]
name = "windows_i686_gnu" name = "windows_i686_gnu"
version = "0.48.0" version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "622a1962a7db830d6fd0a69683c80a18fda201879f0f447f065a3b7467daa241" 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]] [[package]]
name = "windows_i686_msvc" name = "windows_i686_msvc"
version = "0.48.0" version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4542c6e364ce21bf45d69fdd2a8e455fa38d316158cfd43b3ac1c5b1b19f8e00" checksum = "4542c6e364ce21bf45d69fdd2a8e455fa38d316158cfd43b3ac1c5b1b19f8e00"
[[package]]
name = "windows_i686_msvc"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
[[package]] [[package]]
name = "windows_x86_64_gnu" name = "windows_x86_64_gnu"
version = "0.48.0" version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca2b8a661f7628cbd23440e50b05d705db3686f894fc9580820623656af974b1" 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]] [[package]]
name = "windows_x86_64_gnullvm" name = "windows_x86_64_gnullvm"
version = "0.48.0" version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7896dbc1f41e08872e9d5e8f8baa8fdd2677f29468c4e156210174edc7f7b953" 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]] [[package]]
name = "windows_x86_64_msvc" name = "windows_x86_64_msvc"
version = "0.48.0" version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a" checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a"
[[package]]
name = "windows_x86_64_msvc"
version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"

View File

@@ -3,7 +3,7 @@ members = ["crates/*"]
resolver = "2" resolver = "2"
[workspace.package] [workspace.package]
version = "0.1.0" version = "0.1.1"
license = "MIT" license = "MIT"
[workspace.dependencies] [workspace.dependencies]

View File

@@ -62,7 +62,7 @@ async fn example_parallel_map(dataset: Vec<i32>) -> anyhow::Result<()> {
// Update progress // Update progress
let count = processed.fetch_add(1, Ordering::SeqCst) + 1; let count = processed.fetch_add(1, Ordering::SeqCst) + 1;
if count % 10 == 0 { if count.is_multiple_of(10) {
println!(" Processed {}/{} items", count, total); println!(" Processed {}/{} items", count, total);
} }
@@ -127,9 +127,7 @@ async fn example_pipeline(dataset: Vec<i32>) -> anyhow::Result<()> {
stage2 stage2
.add(move |_| async move { .add(move |_| async move {
// Filter even numbers and sum // Filter even numbers and sum
let filtered_sum: i32 = chunk.iter() let filtered_sum: i32 = chunk.iter().filter(|&&x| x % 2 == 0).sum();
.filter(|&&x| x % 2 == 0)
.sum();
tokio::time::sleep(Duration::from_millis(20)).await; tokio::time::sleep(Duration::from_millis(20)).await;
@@ -161,7 +159,11 @@ async fn example_batch_processing(dataset: Vec<i32>) -> anyhow::Result<()> {
.map(|chunk| chunk.to_vec()) .map(|chunk| chunk.to_vec())
.collect(); .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(); let mut workers = Workers::new();
workers.with_limit(3); // Process 3 batches concurrently workers.with_limit(3); // Process 3 batches concurrently
@@ -206,7 +208,11 @@ async fn example_batch_processing(dataset: Vec<i32>) -> anyhow::Result<()> {
let results = results.lock().await; let results = results.lock().await;
let total_processed: usize = results.iter().sum(); 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(()) Ok(())
} }

View File

@@ -0,0 +1,62 @@
#![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()
}

View File

@@ -191,7 +191,10 @@
use std::{future::Future, sync::Arc}; use std::{future::Future, sync::Arc};
use tokio::{sync::Mutex, task::JoinHandle}; use tokio::{
sync::{Mutex, OwnedSemaphorePermit, Semaphore},
task::JoinHandle,
};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
/// Extension traits for common patterns. /// Extension traits for common patterns.
@@ -322,8 +325,7 @@ enum WorkerLimit {
#[default] #[default]
NoLimit, NoLimit,
Amount { Amount {
queue: tokio::sync::mpsc::Sender<()>, done: Arc<tokio::sync::Semaphore>,
done: Arc<Mutex<tokio::sync::mpsc::Receiver<()>>>,
}, },
} }
@@ -331,18 +333,21 @@ impl WorkerLimit {
pub async fn queue_worker(&self) -> WorkerGuard { pub async fn queue_worker(&self) -> WorkerGuard {
match self { match self {
WorkerLimit::NoLimit => {} WorkerLimit::NoLimit => {}
WorkerLimit::Amount { queue, .. } => { WorkerLimit::Amount { done } => {
// Queue work, if the channel is limited, we will block until there is enough room // Queue work, if the channel is limited, we will block until there is enough room
queue let permit = done
.send(()) .clone()
.acquire_owned()
.await .await
.expect("tried to queue work on a closed worker channel"); .expect("to be able to acquire permit");
return WorkerGuard {
_permit: Some(permit),
};
} }
} }
WorkerGuard { WorkerGuard { _permit: None }
limit: self.clone(),
}
} }
} }
@@ -353,24 +358,7 @@ impl WorkerLimit {
/// ///
/// This type is not directly constructible by users. /// This type is not directly constructible by users.
pub struct WorkerGuard { pub struct WorkerGuard {
limit: WorkerLimit, _permit: Option<OwnedSemaphorePermit>,
}
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 { impl Workers {
@@ -542,11 +530,8 @@ impl Workers {
/// # } /// # }
/// ``` /// ```
pub fn with_limit(&mut self, limit: usize) -> &mut Self { pub fn with_limit(&mut self, limit: usize) -> &mut Self {
let (tx, rx) = tokio::sync::mpsc::channel(limit);
self.limit = WorkerLimit::Amount { self.limit = WorkerLimit::Amount {
queue: tx, done: Arc::new(Semaphore::new(limit)),
done: Arc::new(Mutex::new(rx)),
}; };
self self
} }