diff --git a/.hermes/plans/mytheclipse-round21-spec.md b/.hermes/plans/mytheclipse-round21-spec.md new file mode 100644 index 0000000..1b2edf4 --- /dev/null +++ b/.hermes/plans/mytheclipse-round21-spec.md @@ -0,0 +1,48 @@ +# Implementation Spec: Round 21 — CPU Parallel Compute Primitives + +## Goal +Fitur parallel khusus CPU (rayon) yang bounded, panic-isolated, error-aggregated. +Melengkapi `compute()` (single call) dengan batch parallel + fork-join. + +## New API (crates/mytheclipse/src/compute.rs, feature `compute`) + +### 1. `compute_map(items, f) -> Result, ComputeErrors>` +- `pack_items` di rayon compute pool: `par_iter().map(f)` — bounded concurrency + otomatis (rayon work-stealing sizing = CPU cores), ordered output. +- `f: Fn(I::Item) -> Result`: + - item error string → dikumpulkan + - panic per item di-catch (catch_unwind) → jadi error, pool survive +- `ComputeErrors { errors: Vec }` — Display, Error, len, is_empty. + (Tidak pakai AggregateError — feature `compute` harus compile tanpa resiliency.) +- `I: IntoParallelIterator` (rayon) — work langsung di pool, tanpa materialize. + +### 2. `compute_join(a, b) -> Result<(RA, RB), MytheclipseError>` +- `rayon::join` wrapper di compute pool: 2 heavy closures run parallel. +- Panic-isolated (catch_unwind per branch) — pool survive, error jadi + ComputePanic. + +### 3. `compute_par_for_each(items, f) -> Result<(), ComputeErrors>` +- `par_iter().for_each` idiom — fire side-effects parallel di pool. +- Panic isolation per item. + +## Design notes +- Reuse `context().compute_pool` (existing sizing: compute_threads dari + RuntimeConfig / available_parallelism) — konsisten dengan `compute()`. +- `rayon::ThreadPool::install` untuk semua — force run di pool. +- Panic isolation: `std::panic::catch_unwind` + AssertUnwindSafe per item + (sama seperti `compute()` yang sudah proven). +- Bounded = rayon work-stealing — concurrency = pool threads (CPU cores), + bukan item count. Tidak perlu semaphore. + +## Files +- crates/mytheclipse/src/compute.rs (3 fungsi + error type) +- crates/mytheclipse/src/lib.rs (export) +- doctests: compute_map, compute_join, compute_par_for_each +- tests: unit di compute.rs + +## Verification +1. `cargo build --workspace --all-features` — exit 0 +2. `cargo test -p mytheclipse compute --all-features` — 0 FAILED +3. `cargo test -p mytheclipse --doc --all-features` — 0 FAILED +4. `cargo clippy --workspace --all-features` — 0 new +5. spec + commit + push \ No newline at end of file diff --git a/crates/mytheclipse/src/compute.rs b/crates/mytheclipse/src/compute.rs index d00fcbb..5a96f2a 100644 --- a/crates/mytheclipse/src/compute.rs +++ b/crates/mytheclipse/src/compute.rs @@ -2,6 +2,8 @@ use std::panic::{catch_unwind, AssertUnwindSafe}; +use rayon::iter::{IntoParallelIterator, ParallelIterator}; + use crate::context::context; use crate::error::MytheclipseError; @@ -45,6 +47,195 @@ fn panic_payload_to_string(payload: Box) -> String { } } +/// Errors collected while mapping over items on the CPU compute pool. +/// +/// Each entry is one item's error message (from an `Err` return or a panic). +/// Unlike [`crate::aggregate_error::AggregateError`], this lives under the +/// `compute` feature only and carries plain strings, so the compute pool +/// primitives compile without the `resiliency` feature. +#[derive(Debug, Default)] +pub struct ComputeErrors { + errors: Vec, +} + +impl ComputeErrors { + /// Number of items that failed (returned `Err` or panicked). + pub fn len(&self) -> usize { + self.errors.len() + } + + /// `true` when every item succeeded. + pub fn is_empty(&self) -> bool { + self.errors.is_empty() + } + + /// Iterate over the error messages. + pub fn iter(&self) -> impl Iterator { + self.errors.iter().map(String::as_str) + } +} + +impl std::fmt::Display for ComputeErrors { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} compute item(s) failed:", self.errors.len())?; + for e in &self.errors { + write!(f, "\n - {e}")?; + } + Ok(()) + } +} + +impl std::error::Error for ComputeErrors {} + +/// Runs `f` over every item **on the rayon compute pool**, in parallel, +/// returning the results **in input order**. +/// +/// This is the CPU-specific counterpart to [`crate::parallel_map::parallel_map`]: +/// work is distributed across the compute pool's worker threads (sized from +/// [`crate::runtime_auto::RuntimeConfig::auto`] — one per logical core) by +/// rayon's work-stealing scheduler, so concurrency is bounded by the pool and +/// **auto-scales to the host CPU** without a manual `concurrency` parameter. +/// +/// Each item's closure runs inside [`std::panic::catch_unwind`]: a panic is +/// caught and recorded as an error instead of unwinding across the pool. All +/// errors (returned or panicked) are collected into a single +/// [`ComputeErrors`]. +/// +/// ``` +/// use mytheclipse::compute::compute_map; +/// +/// let squares = compute_map(vec![1u32, 2, 3, 4], |x| Ok::<_, String>(x * x)).unwrap(); +/// assert_eq!(squares, vec![1, 4, 9, 16]); +/// +/// // Failures are aggregated, order is preserved: +/// let out = compute_map(vec![1, 2, 3], |x| { +/// if x == 2 { Err("boom".to_string()) } else { Ok(x * 10) } +/// }).unwrap_err(); +/// assert_eq!(out.len(), 1); +/// ``` +pub fn compute_map(items: I, f: F) -> Result, ComputeErrors> +where + I: IntoParallelIterator + Send, + I::Item: Send, + T: Send, + F: Fn(I::Item) -> Result + Send + Sync, +{ + let wrapped = AssertUnwindSafe(f); + let collected: Vec> = context() + .compute_pool + .install(move || { + let f = wrapped; + items + .into_par_iter() + .map(|item| { + catch_unwind(AssertUnwindSafe(|| f(item))) + .unwrap_or_else(|payload| Err(panic_payload_to_string(payload))) + }) + .collect() + }); + + let mut values = Vec::with_capacity(collected.len()); + let mut errors = Vec::new(); + for r in collected { + match r { + Ok(v) => values.push(v), + Err(e) => errors.push(e), + } + } + if errors.is_empty() { + Ok(values) + } else { + Err(ComputeErrors { errors }) + } +} + +/// Runs two heavy closures in parallel on the compute pool, returning both +/// results. +/// +/// This wraps [`rayon::join`] with panic isolation: if either branch panics, +/// its panic is converted into a [`MytheclipseError::ComputePanic`] and the +/// other branch still completes. The pool remains usable afterward. +/// +/// ``` +/// use mytheclipse::compute::compute_join; +/// +/// let (a, b) = compute_join( +/// || (0..1_000_000u64).sum::(), +/// || (1_000_000..2_000_000u64).sum::(), +/// ).unwrap(); +/// assert_eq!(a + b, (0..2_000_000u64).sum::()); +/// ``` +pub fn compute_join( + a: A, + b: B, +) -> Result<(RA, RB), MytheclipseError> +where + A: FnOnce() -> RA + Send, + RA: Send, + B: FnOnce() -> RB + Send, + RB: Send, +{ + let a = AssertUnwindSafe(a); + let b = AssertUnwindSafe(b); + context() + .compute_pool + .install(|| { + let (ra, rb) = rayon::join( + move || catch_unwind(a).map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p))), + move || catch_unwind(b).map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p))), + ); + Ok((ra?, rb?)) + }) +} + +/// Runs `f` over every item on the compute pool in parallel, discarding +/// return values (side effects only), collecting errors and panics. +/// +/// ``` +/// use std::sync::atomic::{AtomicUsize, Ordering}; +/// use std::sync::Arc; +/// use mytheclipse::compute::compute_par_for_each; +/// +/// let count = Arc::new(AtomicUsize::new(0)); +/// let c = Arc::clone(&count); +/// compute_par_for_each(0..100, move |x| { +/// c.fetch_add(x as usize, Ordering::SeqCst); +/// Ok::<_, String>(()) +/// }).unwrap(); +/// assert_eq!(count.load(Ordering::SeqCst), 4950); +/// ``` +pub fn compute_par_for_each(items: I, f: F) -> Result<(), ComputeErrors> +where + I: IntoParallelIterator + Send, + I::Item: Send, + F: Fn(I::Item) -> Result<(), String> + Send + Sync, +{ + let wrapped = AssertUnwindSafe(f); + let collected: Vec> = context() + .compute_pool + .install(move || { + let f = wrapped; + items + .into_par_iter() + .map(|item| { + catch_unwind(AssertUnwindSafe(|| f(item))) + .unwrap_or_else(|payload| Err(panic_payload_to_string(payload))) + }) + .collect() + }); + + if collected.iter().any(|r| r.is_err()) { + Err(ComputeErrors { + errors: collected + .into_iter() + .filter_map(|r| r.err()) + .collect(), + }) + } else { + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; @@ -57,4 +248,74 @@ mod tests { let recovered = compute(|| 1 + 1); assert_eq!(recovered.unwrap(), 2); } + + #[test] + fn compute_map_ordered_and_aggregates_errors() { + let squares = compute_map(vec![1u32, 2, 3, 4], |x| Ok::<_, String>(x * x)).unwrap(); + assert_eq!(squares, vec![1, 4, 9, 16]); + + let err = compute_map(vec![1u32, 2, 3], |x| { + if x == 2 { + Err("boom".to_string()) + } else { + Ok(x * 10) + } + }) + .unwrap_err(); + assert_eq!(err.len(), 1); + assert_eq!(err.iter().next().unwrap(), "boom"); + } + + #[test] + fn compute_map_panic_is_isolated_and_collected() { + let err = compute_map(vec![1u32, 2, 3], |x| { + if x == 2 { + panic!("item panic") + } else { + Ok::<_, String>(x) + } + }) + .unwrap_err(); + assert_eq!(err.len(), 1); + assert!(err.iter().next().unwrap().contains("item panic")); + + // pool still usable + let recovered = compute_map(vec![1u32], |x| Ok::<_, String>(x + 1)).unwrap(); + assert_eq!(recovered, vec![2]); + } + + #[test] + fn compute_join_runs_both_branches() { + let (a, b) = compute_join( + || (0..100_000u64).sum::(), + || (100_000..200_000u64).sum::(), + ) + .unwrap(); + assert_eq!(a + b, (0..200_000u64).sum::()); + } + + #[test] + fn compute_join_panic_is_isolated() { + let a = compute_join(|| panic!("branch a"), || 42u32); + assert!(matches!(a, Err(MytheclipseError::ComputePanic(_)))); + + // pool still usable + let ok = compute_join(|| 1u32, || 2u32).unwrap(); + assert_eq!(ok, (1, 2)); + } + + #[test] + fn compute_par_for_each_runs_all_side_effects() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + let count = Arc::new(AtomicUsize::new(0)); + let c = Arc::clone(&count); + compute_par_for_each(0..100, move |x| { + c.fetch_add(x as usize, Ordering::SeqCst); + Ok::<_, String>(()) + }) + .unwrap(); + assert_eq!(count.load(Ordering::SeqCst), 4950); + } } diff --git a/crates/mytheclipse/src/lib.rs b/crates/mytheclipse/src/lib.rs index 14835b2..b537477 100644 --- a/crates/mytheclipse/src/lib.rs +++ b/crates/mytheclipse/src/lib.rs @@ -109,7 +109,7 @@ pub use error::MytheclipseError; #[cfg(feature = "io")] pub use io::spawn_io; #[cfg(feature = "compute")] -pub use compute::compute; +pub use compute::{compute, compute_join, compute_map, compute_par_for_each, ComputeErrors}; #[cfg(feature = "bg")] pub use bg::spawn_bg;