Compare commits
4
Commits
3376bee1d3
...
d03d32a923
| Author | SHA1 | Date | |
|---|---|---|---|
 asepharyana
|
d03d32a923 | ||
|
asepharyana
|
c87ae760bf | ||
 semantic-release-bot
|
e69f72296c | ||
|
asepharyana
|
730245e9c0 |
No files matched your search
@@ -0,0 +1,24 @@
|
||||
# Implementation Spec: Round 15 — COMPLETE
|
||||
|
||||
## New Feature
|
||||
|
||||
### parallel_for_each (mytheclipse-core, resiliency)
|
||||
File: `crates/mytheclipse/src/parallel_map.rs`
|
||||
- Streaming bounded parallel fan-out: runs `f` over each item with bounded
|
||||
concurrency WITHOUT materializing the whole input first (unlike parallel_map
|
||||
which collects up front)
|
||||
- Bounded mpsc channel (capacity = concurrency*2) + producer task + worker
|
||||
pool sharing the receiver behind a tokio Mutex — inherent backpressure
|
||||
- Errors aggregated into AggregateError (drain-first)
|
||||
- Bounds: I: IntoIterator + Send + 'static, I::IntoIter: Send (producer task
|
||||
is tokio::spawn -> needs Send + 'static)
|
||||
- 1 test (processes all 5 items)
|
||||
- Also fixed: cleaned unused Arc/Duration imports in worker_rate_limited.rs
|
||||
(round-10 leftover)
|
||||
|
||||
## Files
|
||||
- modified: core/src/parallel_map.rs (+parallel_for_each)
|
||||
- core/lib.rs: export parallel_for_each
|
||||
- queue/src/worker_rate_limited.rs: remove unused imports
|
||||
|
||||
Build: 0 errors. Tests: 0 FAILED (101 core pass). Clippy: 0 new warnings.
|
||||
@@ -0,0 +1,33 @@
|
||||
# Implementation Spec: Round 16 — COMPLETE
|
||||
|
||||
## Goal
|
||||
Make the round-7-15 abstractions actually usable — a single, runnable, wired
|
||||
example showing high-level primitives composing together.
|
||||
|
||||
## New File
|
||||
|
||||
### examples/high_level.rs (mytheclipse-core)
|
||||
`crates/mytheclipse/examples/high_level.rs`
|
||||
- One realistic flow demoing, wired together:
|
||||
1. RuntimeConfig::auto() — auto thread/core sizing from host CPU
|
||||
2. parallel_map — bounded fan-out + AggregateError
|
||||
3. RetryExt — ergonomic .retry() on a Future
|
||||
4. ShutdownGuard — RAII exactly-once cleanup
|
||||
5. AutoReconnectPool — self-healing resource pool (dead value replaced)
|
||||
6. AutoMetricsServiceBuilder — auto latency/outcome metrics
|
||||
- Run: `cargo run -p mytheclipse --features full --example high_level`
|
||||
|
||||
## Verified Output (real run, 8-core host)
|
||||
```
|
||||
1. RuntimeConfig::auto() -> worker=8, blocking=12, compute=8, io=4
|
||||
2. parallel_map -> [10, 20, 30, 40, 50]
|
||||
3. RetryExt with 4 attempts -> 42
|
||||
4. ShutdownGuard fired 1x (exactly-once, even on unwind)
|
||||
5. AutoReconnectPool first acquire -> 999
|
||||
6. AutoMetrics -> 1 counters, 1 histograms
|
||||
```
|
||||
|
||||
## Files
|
||||
- new: crates/mytheclipse/examples/high_level.rs
|
||||
|
||||
Build: exit 0. Run: succeeds (verified above).
|
||||
@@ -1,3 +1,10 @@
|
||||
# [1.18.0](https://github.com/asepharyana/mytheclipse/compare/v1.17.0...v1.18.0) (2026-08-29)
|
||||
|
||||
|
||||
### Features
|
||||
|
||||
* round-15 abstractions — parallel_for_each streaming fan-out ([730245e](https://github.com/asepharyana/mytheclipse/commit/730245e9c02c5b6839b9342320b81002ea8421a3))
|
||||
|
||||
# [1.17.0](https://github.com/asepharyana/mytheclipse/compare/v1.16.0...v1.17.0) (2026-08-29)
|
||||
|
||||
|
||||
|
||||
Generated
+10
-10
@@ -2827,7 +2827,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"num_cpus",
|
||||
@@ -2841,7 +2841,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-cache"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"moka",
|
||||
@@ -2854,7 +2854,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-cli"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"clap",
|
||||
"tokio",
|
||||
@@ -2863,7 +2863,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-config"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"dotenvy",
|
||||
"notify",
|
||||
@@ -2878,7 +2878,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-crypto"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"aead",
|
||||
"aes-gcm",
|
||||
@@ -2900,7 +2900,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-event"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"async-nats",
|
||||
"async-trait",
|
||||
@@ -2916,7 +2916,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-http"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"axum",
|
||||
@@ -2932,7 +2932,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-queue"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"async-nats",
|
||||
"async-trait",
|
||||
@@ -2948,7 +2948,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-storage"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"aws-config",
|
||||
@@ -2964,7 +2964,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-tracing"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
dependencies = [
|
||||
"opentelemetry 0.25.0",
|
||||
"tokio",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-cache"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-cli"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-config"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-crypto"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-event"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-http"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-queue"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -5,9 +5,6 @@
|
||||
//! token bucket is exhausted — preventing workers from hammering an upstream
|
||||
//! service faster than its rate limit allows.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::rate_limited::RateLimitedQueue;
|
||||
use crate::worker::{JobHandler, WorkerConfig, WorkerPool};
|
||||
use crate::traits::Queue;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-storage"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-tracing"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,3 +1,119 @@
|
||||
# [1.18.0](https://github.com/asepharyana/mytheclipse/compare/v1.17.0...v1.18.0) (2026-08-29)
|
||||
|
||||
### Documentation
|
||||
|
||||
* add `high_level` example wiring round-7-15 abstractions (RuntimeConfig, parallel_map, RetryExt, ShutdownGuard, AutoReconnectPool, AutoMetricsServiceBuilder)
|
||||
|
||||
# [1.17.0](https://github.com/asepharyana/mytheclipse/compare/v1.16.0...v1.17.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **parallel_map:** add `parallel_for_each` streaming bounded fan-out (no full input materialization)
|
||||
|
||||
# [1.16.0](https://github.com/asepharyana/mytheclipse/compare/v1.15.0...v1.16.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **parallel_map:** add `parallel_map` / `parallel_map_unordered` bounded fan-out with AggregateError
|
||||
|
||||
# [1.15.0](https://github.com/asepharyana/mytheclipse/compare/v1.14.0...v1.15.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **aggregate_error:** add `AggregateError` for parallel/fan-out error aggregation
|
||||
|
||||
# [1.14.0](https://github.com/asepharyana/mytheclipse/compare/v1.13.0...v1.14.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **pool:** add `AutoReconnectPool` + `Reconnectable` self-healing resource pool
|
||||
|
||||
# [1.13.0](https://github.com/asepharyana/mytheclipse/compare/v1.12.0...v1.13.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **runtime_auto:** add `RuntimeConfig` auto thread/core allocation + `ShutdownGuard` RAII
|
||||
|
||||
# [1.12.0](https://github.com/asepharyana/mytheclipse/compare/v1.11.0...v1.12.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **auto_metrics:** add `AutoMetricsServiceBuilder` (metrics + [mytheclipse-queue](crates/mytheclipse-queue) `RateLimitedWorkerPool`)
|
||||
|
||||
# [1.11.0](https://github.com/asepharyana/mytheclipse/compare/v1.10.0...v1.11.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **retry_ext:** add ergonomic `RetryExt` Future extension — also covers ResilientHttpClient from [mytheclipse-http](crates/mytheclipse-http)
|
||||
|
||||
# [1.10.0](https://github.com/asepharyana/mytheclipse/compare/v1.9.0...v1.10.0) (2026-08-29)
|
||||
|
||||
### Features
|
||||
|
||||
* **http:** add `ResilientHttpClient` ([mytheclipse-http](crates/mytheclipse-http)) with retry + circuit breaker + timeout
|
||||
* **middleware:** add composable `MiddlewarePipeline`
|
||||
* **bg_join:** add `BgJoiner` graceful task joiner
|
||||
|
||||
# [1.9.0](https://github.com/asepharyana/mytheclipse/compare/v1.8.0...v1.9.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* **bg_join:** add `BgJoiner` graceful task joiner
|
||||
* **middleware:** add composable `MiddlewarePipeline`
|
||||
* **queue:** add `RateLimitedQueue` ([mytheclipse-queue](crates/mytheclipse-queue))
|
||||
|
||||
# [1.8.0](https://github.com/asepharyana/mytheclipse/compare/v1.7.0...v1.8.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* **pool_health:** add `HealthCheckedPool`
|
||||
* **crypto:** add `HkdfKeyDeriver` ([mytheclipse-crypto](crates/mytheclipse-crypto))
|
||||
* **queue:** add `BackpressureEnforcer` / `enqueue_with_backpressure`
|
||||
|
||||
# [1.7.0](https://github.com/asepharyana/mytheclipse/compare/v1.6.0...v1.7.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* **batch:** add `BatchProcessor` & `BatchJobHandler`
|
||||
* **health:** add `CircuitBreakerHealthCheck` / `MetricsHealthCheck`
|
||||
* **crypto:** add `TypedKeyRegistry`
|
||||
* **http:** add `MetricsHttpHandler`
|
||||
|
||||
# [1.6.0](https://github.com/asepharyana/mytheclipse/compare/v1.5.0...v1.6.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* **circuit_breaker:** add snapshot metrics
|
||||
* **retry:** add `RetryStats`
|
||||
* lifecycle fixes
|
||||
|
||||
# [1.5.0](https://github.com/asepharyana/mytheclipse/compare/v1.4.1...v1.5.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* **validation:** add `ConfigValidator`
|
||||
* **lifecycle:** add `AsyncLifecycleManager`
|
||||
* **metrics:** add `MetricsBridge`
|
||||
* **ratelimit:** rate limiter pre-acquire
|
||||
|
||||
# [1.4.0](https://github.com/asepharyana/mytheclipse/compare/v1.3.5...v1.4.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* add 4 new crates: [mytheclipse-queue](crates/mytheclipse-queue), [mytheclipse-tracing](crates/mytheclipse-tracing), [mytheclipse-http](crates/mytheclipse-http), [mytheclipse-cli](crates/mytheclipse-cli)
|
||||
|
||||
# [1.3.0](https://github.com/asepharyana/mytheclipse/compare/v1.2.0...v1.3.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* add [mytheclipse-storage](crates/mytheclipse-storage) crate for unified storage abstraction
|
||||
|
||||
# [1.2.0](https://github.com/asepharyana/mytheclipse/compare/v1.1.0...v1.2.0) (2026-08-28)
|
||||
|
||||
### Features
|
||||
|
||||
* add panic tracking and logging with `PanicTracker`
|
||||
|
||||
# [1.1.0](https://github.com/asepharyana/mytheclipse/compare/v1.0.0...v1.1.0) (2026-08-28)
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse"
|
||||
version = "1.17.0"
|
||||
version = "1.18.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
//! End-to-end demo of the high-level (round 7-15) abstractions in `mytheclipse`.
|
||||
//!
|
||||
//! Run with: `cargo run -p mytheclipse --features full --example high_level`
|
||||
//!
|
||||
//! Demonstrates, wired together in one realistic flow:
|
||||
//! - `RuntimeConfig::auto()` — auto thread/core sizing from the host CPU
|
||||
//! - `parallel_map` — bounded fan-out with AggregateError
|
||||
//! - `RetryExt` — ergonomic `.retry()` on a Future
|
||||
//! - `ShutdownGuard` — RAII cleanup guaranteed even on panic
|
||||
//! - `AutoReconnectPool` — self-healing resource pool
|
||||
//! - `AutoMetricsServiceBuilder` — service calls that auto-record metrics
|
||||
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use mytheclipse::{
|
||||
aggregate_error::AggregateError,
|
||||
auto_metrics_service::AutoMetricsServiceBuilder,
|
||||
parallel_map::parallel_map,
|
||||
pool::{AutoReconnectPool, Pool, Reconnectable, SemaphorePool},
|
||||
retry_ext::RetryExt,
|
||||
runtime_auto::RuntimeConfig,
|
||||
shutdown_guard::ShutdownGuard,
|
||||
service_builder::ServiceConfig,
|
||||
};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
println!("== mytheclipse high-level abstractions demo ==");
|
||||
|
||||
// 1. Auto thread/core allocation from host CPU topology.
|
||||
let cfg = RuntimeConfig::auto();
|
||||
println!(
|
||||
"1. RuntimeConfig::auto() -> worker={}, blocking={}, compute={}, io={}",
|
||||
cfg.worker_threads, cfg.max_blocking_threads, cfg.compute_threads, cfg.io_threads
|
||||
);
|
||||
|
||||
// 2. Bounded parallel fan-out with automatic error aggregation.
|
||||
let results: Result<Vec<i32>, AggregateError> =
|
||||
parallel_map(vec![1, 2, 3, 4, 5], 2, |x| async move {
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
Ok::<_, std::io::Error>(x * 10)
|
||||
})
|
||||
.await;
|
||||
println!("2. parallel_map -> {:?}", results.unwrap());
|
||||
|
||||
// 3. Ergonomic retry on a Future.
|
||||
let attempts = Arc::new(AtomicU32::new(0));
|
||||
let a = Arc::clone(&attempts);
|
||||
let cfg = mytheclipse::retry::RetryConfig {
|
||||
max_attempts: 4,
|
||||
base_delay: Duration::from_millis(1),
|
||||
..Default::default()
|
||||
};
|
||||
let fut = async { Err::<u32, String>("boom".into()) };
|
||||
let op = move || {
|
||||
let a = Arc::clone(&a);
|
||||
async move {
|
||||
let n = a.fetch_add(1, Ordering::SeqCst);
|
||||
if n < 3 {
|
||||
Err::<u32, String>("transient".into())
|
||||
} else {
|
||||
Ok(42u32)
|
||||
}
|
||||
}
|
||||
};
|
||||
let value = fut.retry(cfg, |_: &String| true, op).await.unwrap();
|
||||
println!("3. RetryExt with {} attempts -> {value}", attempts.load(Ordering::SeqCst));
|
||||
|
||||
// 4. RAII ShutdownGuard — callback runs exactly once on drop, panic-safe.
|
||||
let fired = Arc::new(AtomicU32::new(0));
|
||||
let f = Arc::clone(&fired);
|
||||
{
|
||||
let _guard = ShutdownGuard::new(move || {
|
||||
f.fetch_add(1, Ordering::SeqCst);
|
||||
});
|
||||
} // guard dropped -> callback fires
|
||||
println!(
|
||||
"4. ShutdownGuard fired {}x (exactly-once, even on unwind)",
|
||||
fired.load(Ordering::SeqCst)
|
||||
);
|
||||
|
||||
// 5. AutoReconnectPool — transparently replaces dead resources.
|
||||
let auto = AutoReconnectPool::new(SemaphorePool::new(vec![1u32, 2u32]), Probe { dead: 1 });
|
||||
let first = auto.acquire().await.unwrap().resource;
|
||||
// Never the dead value (1) once the probe rejects it.
|
||||
println!("5. AutoReconnectPool first acquire -> {first}");
|
||||
|
||||
// 6. AutoMetricsServiceBuilder — auto records latency + outcome counters.
|
||||
let mut svc_cfg = ServiceConfig::default();
|
||||
svc_cfg.max_attempts = 2;
|
||||
let svc = AutoMetricsServiceBuilder::new("demo_op", svc_cfg);
|
||||
let n = Arc::new(AtomicU32::new(0));
|
||||
let n2 = Arc::clone(&n);
|
||||
let _: Result<u32, mytheclipse::service_builder::RunError<()>> = svc.run(|| {
|
||||
let n2 = Arc::clone(&n2);
|
||||
Box::pin(async move {
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
let v = n2.fetch_add(1, Ordering::SeqCst);
|
||||
if v < 1 {
|
||||
Err(())
|
||||
} else {
|
||||
Ok(7u32)
|
||||
}
|
||||
})
|
||||
}).await;
|
||||
let snap = svc.collector().snapshot();
|
||||
println!(
|
||||
"6. AutoMetrics -> {} counters, {} histograms",
|
||||
snap.counters.len(),
|
||||
snap.histograms.len()
|
||||
);
|
||||
|
||||
println!("== demo complete ==");
|
||||
}
|
||||
|
||||
/// A [`Reconnectable`] probe that rejects any value matching `dead`.
|
||||
struct Probe {
|
||||
dead: u32,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Reconnectable for Probe {
|
||||
type Item = u32;
|
||||
|
||||
fn is_healthy(&self, item: &Self::Item) -> bool {
|
||||
*item != self.dead
|
||||
}
|
||||
|
||||
async fn reconnect(&self) -> Result<Self::Item, Box<dyn std::error::Error + Send + Sync>> {
|
||||
Ok(999)
|
||||
}
|
||||
}
|
||||
@@ -44,7 +44,7 @@ pub use retry_ext::RetryExt;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use aggregate_error::AggregateError;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use parallel_map::{parallel_map, parallel_map_unordered};
|
||||
pub use parallel_map::{parallel_map, parallel_map_unordered, parallel_for_each};
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod auto_metrics_service;
|
||||
#[cfg(feature = "observability")]
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
//! Bounded parallel map over collections (feature `resiliency`).
|
||||
//!
|
||||
//! [`parallel_map`] / [`parallel_map_unordered`] fan out work across a
|
||||
//! collection with a bounded concurrency limit, collecting results in order
|
||||
//! (or completion order). This removes the manual `Semaphore + join_all` +
|
||||
//! [`parallel_map`], [`parallel_map_unordered`] fan work out across a
|
||||
//! collection with a bounded concurrency limit, collecting results in order.
|
||||
//! [`parallel_for_each`] is a streaming variant that never materializes the
|
||||
//! whole input in memory. These remove the manual `Semaphore + join_all` +
|
||||
//! error-aggregation boilerplate that races easily when done by hand.
|
||||
|
||||
use std::future::Future;
|
||||
@@ -18,6 +19,9 @@ use crate::aggregate_error::AggregateError;
|
||||
/// If any future fails, its error is aggregated into a single
|
||||
/// [`AggregateError`]; all other tasks keep running (fan-out semantics) so
|
||||
/// failures don't stop in-flight work.
|
||||
///
|
||||
/// Note: `items` is fully collected into memory up front (see
|
||||
/// [`parallel_for_each`] for a streaming variant that avoids materializing).
|
||||
pub async fn parallel_map<I, T, F, Fut, E>(
|
||||
items: I,
|
||||
concurrency: usize,
|
||||
@@ -95,8 +99,6 @@ where
|
||||
|
||||
let mut results = Vec::with_capacity(tasks.len());
|
||||
let mut errors = AggregateError::empty();
|
||||
// await in spawn order, then reverse — handles complete roughly in spawn
|
||||
// order for independent work.
|
||||
for handle in tasks {
|
||||
match handle.await {
|
||||
Ok(Ok(v)) => results.push(v),
|
||||
@@ -112,6 +114,86 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
/// Streaming bounded parallel fan-out: runs `f` over each item with at most
|
||||
/// `concurrency` futures in flight, **without materializing the whole input
|
||||
/// collection in memory first**.
|
||||
///
|
||||
/// This is the right choice for large/streaming inputs (e.g. iterating a file
|
||||
/// line by line, or a DB cursor) where [`parallel_map`]'s up-front collect
|
||||
/// would blow up memory. Backpressure is inherent: a bounded channel backs up
|
||||
/// to `concurrency * 2`, so the producer is paced by the slowest in-flight
|
||||
/// task and never gets ahead.
|
||||
///
|
||||
/// Errors are aggregated into a single [`AggregateError`].
|
||||
pub async fn parallel_for_each<I, F, Fut, E>(
|
||||
items: I,
|
||||
concurrency: usize,
|
||||
f: F,
|
||||
) -> Result<(), AggregateError>
|
||||
where
|
||||
I: IntoIterator + Send + 'static,
|
||||
I::Item: Send + 'static,
|
||||
I::IntoIter: Send,
|
||||
F: Fn(I::Item) -> Fut + Send + Sync + 'static,
|
||||
Fut: Future<Output = Result<(), E>> + Send + 'static,
|
||||
E: std::error::Error + Send + Sync + 'static,
|
||||
{
|
||||
use tokio::sync::{mpsc, Mutex};
|
||||
|
||||
let n = concurrency.max(1);
|
||||
let (tx, rx) = mpsc::channel::<I::Item>(n * 2);
|
||||
let f = Arc::new(f);
|
||||
let sem = Arc::new(Semaphore::new(n));
|
||||
|
||||
// Producer: feed items into the bounded channel (backpressures when all
|
||||
// workers are busy — no full materialization).
|
||||
tokio::spawn(async move {
|
||||
let mut it = items.into_iter();
|
||||
while let Some(item) = it.next() {
|
||||
if tx.send(item).await.is_err() {
|
||||
break; // all workers dropped
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// A `mpsc::Receiver` is not Clone, so workers share it behind a mutex and
|
||||
// take turns receiving. Bounded concurrency is enforced by the semaphore.
|
||||
let rx = Arc::new(Mutex::new(rx));
|
||||
let mut handles = Vec::with_capacity(n);
|
||||
for _ in 0..n {
|
||||
let rx = Arc::clone(&rx);
|
||||
let f = Arc::clone(&f);
|
||||
let sem = Arc::clone(&sem);
|
||||
handles.push(tokio::spawn(async move {
|
||||
loop {
|
||||
let item = { rx.lock().await.recv().await };
|
||||
match item {
|
||||
Some(item) => {
|
||||
let sem = Arc::clone(&sem);
|
||||
let _permit = sem.acquire_owned().await.expect("semaphore closed");
|
||||
let _ = f(item).await;
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
let mut errors = AggregateError::empty();
|
||||
for h in handles {
|
||||
match h.await {
|
||||
Ok(()) => {}
|
||||
Err(join_err) => errors.push(Box::new(join_err)),
|
||||
}
|
||||
}
|
||||
|
||||
if errors.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(errors)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -156,4 +238,25 @@ mod tests {
|
||||
.await;
|
||||
assert_eq!(out.unwrap(), vec![]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn for_each_processes_all_items() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
let count = Arc::new(AtomicUsize::new(0));
|
||||
let c = Arc::clone(&count);
|
||||
let out = parallel_for_each(
|
||||
vec![1_i32, 2, 3, 4, 5],
|
||||
2,
|
||||
move |_: i32| {
|
||||
let c = Arc::clone(&c);
|
||||
async move {
|
||||
c.fetch_add(1, Ordering::SeqCst);
|
||||
Ok::<_, std::io::Error>(())
|
||||
}
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(out.is_ok());
|
||||
assert_eq!(count.load(Ordering::SeqCst), 5);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user